推式来源的工步与循环改为从采样段推导

yeg

5 天以前 99935242ecccd94e76978e2b71cbc46b87d31dc6

 battery-business/src/test/java/com/battery/business/service/impl/FakeIngestLanding.java

@@ -32,6 +32,22 @@
        return 1;
    }

    /** 拓扑是声明：撞键即刷新声明内容，不追加新行（与生产 SQL 的 upsert 语义一致）。 */
    @Override
    public int upsertTopology(TDeviceIngestRecord record)
    {
        int index = -1;
        for (int i = 0; i < rows.size(); i++)
            if (same(rows.get(i), record)) index = i;
        if (index < 0)
        {
            rows.add(record);
            return 1;
        }
        rows.set(index, record);
        return 1;
    }

    @Override
    public String selectIngestRecordHashByKey(String adapterCode, String payloadKind, String recordKey)
    {
@@ -59,6 +75,7 @@
            TDeviceIngestChannel channel = new TDeviceIngestChannel();
            channel.setDevtype(row.getDevtype()); channel.setDevid(row.getDevid());
            channel.setSubdevid(row.getSubdevid()); channel.setChlid(row.getChlid());
            channel.setVendorChannelNo(row.getVendorChannelNo());
            channel.setLastDataAt(rows.stream().filter(sample -> isSample(sample, adapterCode, row))
                    .map(TDeviceIngestRecord::getDeviceTimeUtc).filter(java.util.Objects::nonNull)
                    .max(Date::compareTo).orElse(null));
@@ -80,6 +97,17 @@
    }

    @Override
    public List<TDeviceIngestRecord> selectSamplesOfSource(String adapterCode, int devtype, int devid, int subdevid,
            int chlid, String sourceRef)
    {
        return samples(adapterCode, devtype, devid, subdevid, chlid).stream()
                .filter(row -> sourceRef.equals(row.getSourceRef()))
                .sorted(java.util.Comparator.comparing(TDeviceIngestRecord::getDeviceTimeUtc)
                        .thenComparing(TDeviceIngestRecord::getRecordSequence))
                .toList();
    }

    @Override
    public List<TDeviceIngestRecord> selectSamplesSince(String adapterCode, int devtype, int devid, int subdevid,
            int chlid, Date watermarkTime, long watermarkSequence, int limit)
    {