shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: make event file store optional

Browse Source
This commit is contained in:
lingniu
2026-07-01 07:51:45 +08:00
parent 51b1dd4174
commit 56a6c01f95
32 changed files with 139 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStore.java
View File

@@ -3,6 +3,9 @@ package com.lingniu.ingest.eventfilestore;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import java.io.IOException;
import java.nio.file.Files;

97
modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileQuery.java
View File

@@ -1,97 +0,0 @@
package com.lingniu.ingest.eventfilestore;
import com.lingniu.ingest.api.ProtocolId;
import java.time.Instant;
import java.time.LocalDate;
/**
* 文件历史库查询条件。
*
* <p>{@code dateFrom/dateTo} 用于定位分区目录，{@code eventTimeFrom/eventTimeTo}
* 用于在分区内做精确时间过滤。这样既支持按天快速裁剪，也支持接口精确到秒的查询。
*/
public record EventFileQuery(
ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Instant eventTimeFrom,
Instant eventTimeTo,
Order order,
int limit,
String vin,
String eventType,
Instant cursorEventTime,
Instant cursorIngestTime,
String cursorEventId
) {
public enum Order {
ASC,
DESC
}
public EventFileQuery {
if (protocol == null) {
throw new IllegalArgumentException("protocol must not be null");
}
if (dateFrom == null || dateTo == null) {
throw new IllegalArgumentException("date range must not be null");
}
if (dateTo.isBefore(dateFrom)) {
throw new IllegalArgumentException("dateTo must not be before dateFrom");
}
if (eventTimeFrom != null && eventTimeTo != null && eventTimeTo.isBefore(eventTimeFrom)) {
throw new IllegalArgumentException("eventTimeTo must not be before eventTimeFrom");
}
// 归一化可选字段，避免空字符串进入 SQL 谓词或文件分区路径。
order = order == null ? Order.ASC : order;
limit = limit <= 0 ? 100 : limit;
vin = vin == null || vin.isBlank() ? null : vin.trim();
eventType = eventType == null || eventType.isBlank() ? null : eventType.trim();
cursorEventId = cursorEventId == null || cursorEventId.isBlank() ? null : cursorEventId.trim();
if ((cursorEventTime == null || cursorIngestTime == null || cursorEventId == null)
&& !(cursorEventTime == null && cursorIngestTime == null && cursorEventId == null)) {
throw new IllegalArgumentException("cursor eventTime, ingestTime and eventId must be provided together");
}
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Instant eventTimeFrom,
Instant eventTimeTo,
Order order,
int limit,
String vin,
String eventType) {
this(protocol, dateFrom, dateTo, eventTimeFrom, eventTimeTo, order, limit,
vin, eventType, null, null, null);
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Order order,
int limit,
String vin,
String eventType) {
this(protocol, dateFrom, dateTo, null, null, order, limit, vin, eventType);
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Order order,
int limit) {
this(protocol, dateFrom, dateTo, order, limit, null, null);
}
public EventFileQuery(ProtocolId protocol,
LocalDate dateFrom,
LocalDate dateTo,
Order order,
int limit,
String vin) {
this(protocol, dateFrom, dateTo, order, limit, vin, null);
}
}

48
modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileRecord.java
View File

@@ -1,48 +0,0 @@
package com.lingniu.ingest.eventfilestore;
import com.lingniu.ingest.api.ProtocolId;
import java.time.Instant;
import java.util.Map;
/**
* 文件型明细库的一条可查询记录。
*
* <p>{@code payloadJson} 保存统一 telemetry snapshot JSON；公共列只保留排序、分区、
* 追溯和通用展示需要的最小字段。
*
* <p>对 GB32960 来说，{@code rawArchiveUri} 是最关键的追溯列：通用查询可以直接展示
* payloadJson，专用全字段查询则通过 rawArchiveUri 回读原始包重新解码。
*/
public record EventFileRecord(
String eventId,
ProtocolId protocol,
String eventType,
String vin,
Instant eventTime,
Instant ingestTime,
String rawArchiveUri,
Map<String, String> metadata,
String payloadJson
) {
public EventFileRecord {
if (eventId == null || eventId.isBlank()) {
throw new IllegalArgumentException("eventId must not be blank");
}
if (protocol == null) {
throw new IllegalArgumentException("protocol must not be null");
}
if (eventTime == null) {
throw new IllegalArgumentException("eventTime must not be null");
}
if (ingestTime == null) {
throw new IllegalArgumentException("ingestTime must not be null");
}
// 允许 eventType/vin/rawArchiveUri 为空字符串，便于保存平台登录、RAW 索引等非车辆事件。
eventType = eventType == null ? "" : eventType;
vin = vin == null ? "" : vin;
rawArchiveUri = rawArchiveUri == null ? "" : rawArchiveUri;
metadata = metadata == null ? Map.of() : Map.copyOf(metadata);
payloadJson = payloadJson == null || payloadJson.isBlank() ? "{}" : payloadJson;
}
}

32
modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileStore.java
View File

@@ -1,32 +0,0 @@
package com.lingniu.ingest.eventfilestore;
import java.io.IOException;
import java.util.List;
/**
* 历史明细文件库抽象。
*
* <p>写入侧只接受已经标准化的 {@link EventFileRecord}；具体实现可以落 Parquet、维护 DuckDB
* sidecar 索引，或在测试中用内存实现。32960 专用 snapshot 查询会先用这里按 VIN/日期找到
* rawArchiveUri，再回读原始 .bin 解码完整字段。
*/
public interface EventFileStore {
/** 单条追加的便捷方法，最终仍走批量写入路径，保证实现只维护一种落盘语义。 */
default void append(EventFileRecord record) throws IOException {
appendAll(List.of(record));
}
void appendAll(List<EventFileRecord> records) throws IOException;
/** 按协议、日期、VIN、事件类型等条件查询标准化历史记录。 */
List<EventFileRecord> query(EventFileQuery query) throws IOException;
/**
* 按 rawArchiveUri 回查索引记录。默认实现返回 null，允许轻量测试实现不维护该索引。
* 生产 DuckDB/Parquet 实现必须覆盖，用于 snapshots 的 sourceFrames 反查。
*/
default EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
return null;
}
}

2
modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileStoreSink.java
View File

@@ -5,6 +5,8 @@ import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.api.event.TelemetrySnapshot;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.event.VehicleEventTelemetrySnapshotMapper;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.api.sink.EventSink;
import java.io.IOException;

2
modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java
View File

@@ -2,8 +2,8 @@ package com.lingniu.ingest.eventfilestore.config;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.eventfilestore.DuckDbParquetEventFileStore;
import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.eventfilestore.EventFileStoreSink;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;

5
modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStoreTest.java
View File

@@ -1,7 +1,10 @@
package com.lingniu.ingest.eventfilestore;
import com.lingniu.ingest.api.ProtocolId;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.lingniu.ingest.api.ProtocolId;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

3
modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/EventFileStoreSinkTest.java
View File

@@ -7,6 +7,9 @@ import com.lingniu.ingest.api.event.LocationPayload;
import com.lingniu.ingest.api.event.RawArchiveKeys;
import com.lingniu.ingest.api.event.RealtimePayload;
import com.lingniu.ingest.api.event.VehicleEvent;
import com.lingniu.ingest.api.history.EventFileQuery;
import com.lingniu.ingest.api.history.EventFileRecord;
import com.lingniu.ingest.api.history.EventFileStore;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

2
modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java
View File

@@ -1,6 +1,6 @@
package com.lingniu.ingest.eventfilestore.config;
import com.lingniu.ingest.eventfilestore.EventFileStore;
import com.lingniu.ingest.api.history.EventFileStore;
import com.lingniu.ingest.eventfilestore.EventFileStoreSink;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;