From 7ce64e99636d8f5f00c77d3d22c965a651671dfe Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 11:04:11 +0800 Subject: [PATCH] refactor: remove legacy event file store --- CHANGELOG.md | 3 + DECISIONS.md | 12 +- README.md | 7 +- docs/module-data-flow.html | 13 +- docs/target-architecture.md | 12 +- .../historyapp/MavenModuleProfileTest.java | 81 +-- modules/sinks/event-file-store/pom.xml | 61 -- .../DuckDbParquetEventFileStore.java | 579 ------------------ .../eventfilestore/EventFileStoreSink.java | 261 -------- .../EventFileStoreAutoConfiguration.java | 68 -- .../config/EventFileStoreProperties.java | 97 --- ...ot.autoconfigure.AutoConfiguration.imports | 1 - .../DuckDbParquetEventFileStoreTest.java | 319 ---------- .../EventFileStoreSinkTest.java | 157 ----- .../EventFileStoreAutoConfigurationTest.java | 44 -- pom.xml | 21 - 16 files changed, 61 insertions(+), 1675 deletions(-) delete mode 100644 modules/sinks/event-file-store/pom.xml delete mode 100644 modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStore.java delete mode 100644 modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileStoreSink.java delete mode 100644 modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java delete mode 100644 modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java delete mode 100644 modules/sinks/event-file-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports delete mode 100644 modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStoreTest.java delete mode 100644 modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/EventFileStoreSinkTest.java delete mode 100644 modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 7d5b33be..1d3b030e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,6 +61,9 @@ - **可观测性**：Micrometer/Actuator 装配（`observability`）。 - **终端控制命令网关**（`command-gateway`）。 +### Removed +- 旧文件型事件索引模块和 DuckDB 依赖；历史查询统一收敛到 TDengine 与 `archive://...` 原始报文引用。 + ### Fixed —— GB/T 32960 应答帧时间字段 - `Gb32960MessageDecoder` 对 `VEHICLE_LOGIN` / `VEHICLE_LOGOUT` / `PLATFORM_LOGIN` / `PLATFORM_LOGOUT` 命令预解析 body 首 6B 采集时间，写入 diff --git a/DECISIONS.md b/DECISIONS.md index 30e4a9e4..37bb8bf4 100644 --- a/DECISIONS.md +++ b/DECISIONS.md @@ -100,11 +100,11 @@ 3. 默认生产链路继续以 `sink-archive`、Kafka raw topic、TDengine raw_frames 为准。 - **Consequences**: 默认构建少一个未部署模块；本地 raw archive 原型仍可通过 `-Poptional-raw-archive-store` 保留和验证。 -## ADR-015 event-file-store 兼容索引：Optional only +## ADR-015 文件型事件索引：Removed - **Status**: Accepted -- **Context**: 默认历史查询已收敛到 TDengine `raw_frames`、`vehicle_locations` 和按需解码；旧 Parquet/DuckDB `event-file-store` 只保留为兼容索引实现，不应把 DuckDB/Parquet 路径拖进默认生产 reactor。 +- **Context**: 默认历史查询已收敛到 TDengine `raw_frames`、`vehicle_locations` 和按需解码；旧文件型索引会增加一套无生产部署的查询和依赖边界。 - **Decision**: - 1. `event-file-store` 不进入默认 Maven reactor。 - 2. event-file-store 仅保留在 `optional-event-file-store` profile，需要兼容排查旧索引时显式构建。 - 3. `EventFileStore`、`EventFileRecord`、`EventFileQuery` 轻量契约归属 `ingest-api`，history 服务只依赖契约，不依赖兼容实现。 -- **Consequences**: 默认构建面少一个 DuckDB/Parquet 兼容模块；旧索引实现仍可通过 `-Poptional-event-file-store` 单独验证。 + 1. 删除旧文件型事件索引模块和 profile。 + 2. 父 POM 不再管理旧索引驱动依赖。 + 3. 历史查询和 RAW 回放统一以 TDengine + `archive://...` 引用为准。 +- **Consequences**: 默认构建面和可选构建面都更小；需要历史查询时只维护 TDengine 一条路径。 diff --git a/README.md b/README.md index 16925080..ff8b1606 100644 --- a/README.md +++ b/README.md @@ -49,8 +49,7 @@ lingniu-vehicle-ingest/ │ │ └── inbound-xinda-push/ 信达 Push 接入（废弃兼容模块，仅 -Plegacy-xinda 显式构建） │ ├── sinks/ │ │ ├── sink-kafka/ Kafka producer + Protobuf Envelope -│ │ ├── sink-archive/ 原始报文冷存 -│ │ └── event-file-store/ 可选兼容索引（旧 Parquet/DuckDB 路径，仅 -Poptional-event-file-store 显式构建） +│ │ └── sink-archive/ 原始报文冷存 │ ├── services/ │ │ ├── event-history-service/ Kafka 全字段事件消费 + 历史查询 │ │ ├── vehicle-state-service/ Kafka 全字段事件消费 + Redis 热状态查询（optional-latest-state profile） @@ -84,8 +83,6 @@ Command Gateway/JT1078 下行与音视频信令不在默认 Maven reactor 中； Redis 最新状态查询不在默认 Maven reactor 中；需要时使用 `-Poptional-latest-state` 构建 `vehicle-state-service`。 -旧 Parquet/DuckDB 事件索引不在默认 Maven reactor 中；需要兼容排查时使用 `-Poptional-event-file-store` 构建 `event-file-store`。 - ## 迁移说明 本项目是 `lingniu-vehicle-data-reception` 的 v2 重构，采用 strangler fig 渐进式迁移，旧项目保留只读参考。迁移路径与决策参见 `../REFRACTOR_PLAN.md`。 @@ -102,4 +99,4 @@ Redis 最新状态查询不在默认 Maven reactor 中；需要时使用 `-Popti 2. **协议即插拔**：每个 `protocol-*` / `inbound-*` 都可独立开关（`lingniu.ingest..enabled`） 3. **顺序保证**：同一 VIN 严格有序（Disruptor hash + Kafka 分区 key） 4. **幂等消费**：Envelope 带 `eventId`，下游去重 -5. **原始可回放**：协议接入会把 `rawArchiveKey/rawArchiveUri` 追加到事件 metadata；Kafka Envelope、TDengine `raw_frames` 和导出都使用 `archive://...` 逻辑 URI 追溯原始 bytes，实际文件由 `sink-archive` 管理；`event-file-store` 仅作为旧 Parquet/DuckDB 查询索引兼容路径，通过 `optional-event-file-store` profile 显式构建，不是默认生产历史库 +5. **原始可回放**：协议接入会把 `rawArchiveKey/rawArchiveUri` 追加到事件 metadata；Kafka Envelope、TDengine `raw_frames` 和导出都使用 `archive://...` 逻辑 URI 追溯原始 bytes，实际文件由 `sink-archive` 管理 diff --git a/docs/module-data-flow.html b/docs/module-data-flow.html index e7d29739..32de4e8c 100644 --- a/docs/module-data-flow.html +++ b/docs/module-data-flow.html @@ -320,7 +320,6 @@ flowchart TB mq["sink-kafka
VehicleEvent 到 Protobuf Envelope 到 Kafka"] archive["sink-archive
RawArchive 到本地/S3/OSS 冷存"] tdengineStore["tdengine-history-store
TDengine raw_frames + locations"] - fileStore["event-file-store
Parquet/DuckDB 兼容可选"] end subgraph services["消费服务 modules/services"] @@ -371,7 +370,6 @@ flowchart TB analyticsApp --> stat mq -.可选热状态.-> state history --> tdengineStore - history -.兼容可选.-> fileStore gateway --> session session --> jt808 @@ -391,7 +389,7 @@ flowchart TB class gb,jt808,jt1078,jsatl12,mqtt,mqttProfile,gbApp,jtApp,yutongApp,historyApp,analyticsApp entry; class api,registry,dispatcher,bus,identity core; - class mq,archive,tdengineStore,fileStore sink; + class mq,archive,tdengineStore sink; class gateway,session command; class codec,obs support; @@ -591,13 +589,6 @@ flowchart LR 可回放原始文件、archive://... 逻辑引用。 问题排查合规留痕 - - modules/sinks/event-file-store - 兼容文件型明细库，使用 Parquet 存储解析后的事件明细和 RawArchive 可查询索引，使用 DuckDB 按日期范围正序/倒序查询；默认生产部署不启用，仅用于旧低层记录接口或本地调查场景。 - VehicleEvent 或 EventFileRecord。 - protocol/date 分区的 Parquet 文件和兼容查询结果。 - 兼容查询默认关闭 - modules/sinks/tdengine-history-store 生产历史库边界，负责 TDengine schema、raw_frames/location/telemetry field 行映射、批量写入和分页查询语句；raw_frames 保留完整 payloadJson.parsed，位置表保持轻量字段并通过 rawUri 关联原始帧。 @@ -721,7 +712,7 @@ flowchart LR 明细展示和导出 vehicle-history-app + tdengine-history-store - 生产查询通过 TDengine raw_frames、位置表和 rawUri 关联完成；Parquet/DuckDB 仅作为兼容文件型明细库，不进入默认热路径。 + 生产查询通过 TDengine raw_frames、位置表和 rawUri 关联完成；不再维护文件型事件索引旁路。 diff --git a/docs/target-architecture.md b/docs/target-architecture.md index 7148e748..e98abf31 100644 --- a/docs/target-architecture.md +++ b/docs/target-architecture.md @@ -41,7 +41,7 @@ is explicitly deployed. - Ingest apps do not write business tables. - Ingest apps do not serve history query APIs. - Redis is not the long-term historical store. -- `event-file-store` Parquet/DuckDB is not the default history store. +- File-based event indexes are not part of the current build surface. - `telemetry_fields` parsing is not handled by this project when a separate field parsing service owns that job; the `telemetry_fields` query API is also disabled by default unless `TDENGINE_TELEMETRY_FIELDS_ENABLED=true`. @@ -111,9 +111,8 @@ Production default: - `EVENT_FILE_STORE_ENABLED=false`. - `TDENGINE_TELEMETRY_FIELDS_ENABLED=false`. -`event-file-store` remains an optional compatibility path for old low-level -record APIs or local investigations. It should not be treated as the primary -history backend. +File-based event indexes have been removed; this app should keep history +queries on TDengine and raw replay on `archive://...` references. ### Analytics App @@ -192,10 +191,7 @@ flowchart LR through `optional-latest-state`. - raw-archive-store is absent from the default Maven reactor and available through `optional-raw-archive-store`. -- event-file-store is absent from the default Maven reactor and available - through `optional-event-file-store`. -- History APIs read from TDengine by default; Parquet/DuckDB is optional - compatibility only. +- History APIs read from TDengine; file-based event indexes are not maintained. - Raw frame queries can return complete parsed JSON through `payloadJson.parsed`. - Location queries page over compact rows and reference raw records instead of duplicating full raw JSON. diff --git a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java index e17f0594..62b0632b 100644 --- a/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java +++ b/modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/MavenModuleProfileTest.java @@ -64,19 +64,27 @@ class MavenModuleProfileTest { } @Test - void eventFileStoreCompatibilityPathIsOptionalAndOutsideDefaultProductionReactor() throws Exception { + void eventFileStoreCompatibilityPathIsRemovedFromProductionStack() throws Exception { Document pom = rootPom(); - Document eventHistoryServicePom = modulePom("modules/services/event-history-service/pom.xml"); + String readme = Files.readString(repositoryRoot().resolve("README.md")); assertThat(defaultModules(pom)) .doesNotContain("modules/sinks/event-file-store"); - assertThat(profileModules(pom, "optional-event-file-store")) - .containsExactly("modules/sinks/event-file-store"); - - assertThat(hasDependency(eventHistoryServicePom, "com.lingniu.ingest", "event-file-store")) - .as("event-history-service should depend only on the lightweight history contract") + .as("DuckDB/Parquet event-file-store has been replaced by TDengine raw history") + .isEmpty(); + assertThat(Files.exists(repositoryRoot().resolve("modules/sinks/event-file-store"))) .isFalse(); + assertThat(rootDependencyManagementArtifacts(pom)) + .doesNotContain("event-file-store"); + assertThat(hasProjectProperty(pom, "duckdb.version")) + .isFalse(); + assertThat(hasDependency(pom, "org.duckdb", "duckdb_jdbc")) + .isFalse(); + assertThat(readme) + .doesNotContain("event-file-store") + .doesNotContain("DuckDB") + .doesNotContain("Parquet"); } @Test @@ -124,7 +132,6 @@ class MavenModuleProfileTest { "protocol-jsatl12", "vehicle-state-service", "raw-archive-store", - "event-file-store", "command-gateway", "inbound-xinda-push", "xinda-push-app"); @@ -138,7 +145,7 @@ class MavenModuleProfileTest { assertThat(profileDependencyManagementArtifacts(pom, "optional-raw-archive-store")) .containsExactly("raw-archive-store"); assertThat(profileDependencyManagementArtifacts(pom, "optional-event-file-store")) - .containsExactly("event-file-store"); + .isEmpty(); assertThat(profileDependencyManagementArtifacts(pom, "legacy-xinda")) .containsExactly("inbound-xinda-push", "xinda-push-app"); } @@ -165,7 +172,6 @@ class MavenModuleProfileTest { .contains("session-core/ 设备会话 + 鉴权 + Token + Redis SessionStore") .contains("protocol-jt1078/ JT/T 1078（optional-command-gateway profile") .contains("protocol-jsatl12/ 苏标主动安全报警附件（optional-attachments profile）") - .contains("event-file-store/ 可选兼容索引（旧 Parquet/DuckDB 路径，仅 -Poptional-event-file-store 显式构建）") .contains("vehicle-state-service/ Kafka 全字段事件消费 + Redis 热状态查询（optional-latest-state profile）") .contains("command-gateway/ 可选 HTTP → 设备下行命令（optional-command-gateway profile）") .contains("vehicle-history-app/ TDengine 历史查询 + RAW JSON") @@ -175,6 +181,9 @@ class MavenModuleProfileTest { .doesNotContain("HiveMQ MQTT Client") .doesNotContain("Caffeine 内存降级") .doesNotContain("Redis/Memory SessionStore") + .doesNotContain("event-file-store") + .doesNotContain("DuckDB") + .doesNotContain("Parquet") .doesNotContain("Parquet + DuckDB 文件型明细库") .doesNotContain("车辆状态/日统计消费"); } @@ -228,7 +237,8 @@ class MavenModuleProfileTest { .contains("JSATL12 仅保留在 `optional-attachments` profile") .contains("vehicle-state-service 仅保留在 `optional-latest-state` profile") .contains("raw-archive-store 仅保留在 `optional-raw-archive-store` profile") - .contains("event-file-store 仅保留在 `optional-event-file-store` profile") + .doesNotContain("event-file-store") + .doesNotContain("optional-event-file-store") .doesNotContain("## ADR-004 信达 Push：保留模块但彻底重写\n- **Status**: Accepted") .doesNotContain("Spring Boot 3.4.x"); } @@ -249,12 +259,15 @@ class MavenModuleProfileTest { } @Test - void duckDbDriverDoesNotLeakFromCompatibilityStoreIntoProductionApps() throws Exception { - Document eventFileStorePom = modulePom("modules/sinks/event-file-store/pom.xml"); + void duckDbAndParquetAreNotPartOfTheCurrentBuildSurface() throws Exception { + Document pom = rootPom(); - assertThat(dependencyOptional(eventFileStorePom, "org.duckdb", "duckdb_jdbc")) - .as("event-file-store is a compatibility path; DuckDB must not be transitive") - .isTrue(); + assertThat(hasProjectProperty(pom, "duckdb.version")) + .isFalse(); + assertThat(hasDependency(pom, "org.duckdb", "duckdb_jdbc")) + .isFalse(); + assertThat(Files.exists(repositoryRoot().resolve("modules/sinks/event-file-store"))) + .isFalse(); } private static Document rootPom() throws Exception { @@ -276,6 +289,11 @@ class MavenModuleProfileTest { return property.getTextContent().trim(); } + private static boolean hasProjectProperty(Document pom, String name) { + Element properties = firstDirectChild(pom.getDocumentElement(), "properties"); + return properties != null && firstDirectChild(properties, name) != null; + } + private static List defaultModules(Document pom) { Element project = pom.getDocumentElement(); return directModules(firstDirectChild(project, "modules")); @@ -359,31 +377,20 @@ class MavenModuleProfileTest { return List.copyOf(artifacts); } - private static boolean dependencyOptional(Document pom, String groupId, String artifactId) { - Element dependencies = firstDirectChild(pom.getDocumentElement(), "dependencies"); - if (dependencies == null) { + private static boolean hasDependency(Document pom, String groupId, String artifactId) { + Element owner = pom.getDocumentElement(); + Element dependencyManagement = firstDirectChild(owner, "dependencyManagement"); + Element dependencies = firstDirectChild(owner, "dependencies"); + if (hasDependencyIn(dependencies, groupId, artifactId)) { + return true; + } + if (dependencyManagement == null) { return false; } - NodeList children = dependencies.getChildNodes(); - for (int i = 0; i < children.getLength(); i++) { - if (!(children.item(i) instanceof Element dependency) || !"dependency".equals(dependency.getTagName())) { - continue; - } - Element currentGroupId = firstDirectChild(dependency, "groupId"); - Element currentArtifactId = firstDirectChild(dependency, "artifactId"); - if (currentGroupId != null - && currentArtifactId != null - && groupId.equals(currentGroupId.getTextContent().trim()) - && artifactId.equals(currentArtifactId.getTextContent().trim())) { - Element optional = firstDirectChild(dependency, "optional"); - return optional != null && Boolean.parseBoolean(optional.getTextContent().trim()); - } - } - return false; + return hasDependencyIn(firstDirectChild(dependencyManagement, "dependencies"), groupId, artifactId); } - private static boolean hasDependency(Document pom, String groupId, String artifactId) { - Element dependencies = firstDirectChild(pom.getDocumentElement(), "dependencies"); + private static boolean hasDependencyIn(Element dependencies, String groupId, String artifactId) { if (dependencies == null) { return false; } diff --git a/modules/sinks/event-file-store/pom.xml b/modules/sinks/event-file-store/pom.xml deleted file mode 100644 index a8a21334..00000000 --- a/modules/sinks/event-file-store/pom.xml +++ /dev/null @@ -1,61 +0,0 @@ - - - 4.0.0 - - com.lingniu.ingest - lingniu-vehicle-ingest - 0.1.0-SNAPSHOT - ../../../pom.xml - - event-file-store - event-file-store - 车辆事件文件型明细库：Parquet 存储 + DuckDB 时间顺序查询。 - - - - com.lingniu.ingest - ingest-api - - - org.springframework.boot - spring-boot-starter - - - org.springframework.boot - spring-boot-autoconfigure - - - org.springframework.boot - spring-boot-configuration-processor - true - - - org.duckdb - duckdb_jdbc - true - - - com.fasterxml.jackson.core - jackson-databind - - - com.fasterxml.jackson.datatype - jackson-datatype-jsr310 - - - org.junit.jupiter - junit-jupiter - test - - - org.assertj - assertj-core - test - - - org.springframework.boot - spring-boot-starter-test - test - - - diff --git a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStore.java b/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStore.java deleted file mode 100644 index de35a442..00000000 --- a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStore.java +++ /dev/null @@ -1,579 +0,0 @@ -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; -import java.nio.file.Path; -import java.nio.file.StandardCopyOption; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.PreparedStatement; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.time.Instant; -import java.time.LocalDate; -import java.time.ZoneId; -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.UUID; -import java.util.stream.Stream; - -/** - * 基于 Parquet 文件 + DuckDB sidecar 索引的历史事件库。 - * - *

写入时按 {@code protocol/date/vehicle/header} 分区保存 Parquet，便于按单车、按天直接读取。 - * 同时维护一个轻量 DuckDB 索引文件 {@code events.duckdb}，用于无 VIN 查询、rawArchiveUri - * 精确定位和索引重建。设计目标是把 32960 查询的热路径限制在“单车若干天的文件”内。 - */ -public final class DuckDbParquetEventFileStore implements EventFileStore { - - private static final TypeReference> STRING_MAP = - new TypeReference<>() {}; - private static final String HEADER_VERSION = "event-records-v1"; - private static final int VEHICLE_BUCKETS = 64; - - private final Path root; - private final Path indexPath; - private final ZoneId partitionZone; - private final ObjectMapper objectMapper; - private final boolean writeParquetEnabled; - private volatile boolean indexInitialized; - - public DuckDbParquetEventFileStore(Path root, ZoneId partitionZone) { - this(root, partitionZone, new ObjectMapper()); - } - - public DuckDbParquetEventFileStore(Path root, ZoneId partitionZone, ObjectMapper objectMapper) { - this(root, partitionZone, objectMapper, true); - } - - public DuckDbParquetEventFileStore(Path root, - ZoneId partitionZone, - ObjectMapper objectMapper, - boolean writeParquetEnabled) { - if (root == null) { - throw new IllegalArgumentException("root must not be null"); - } - this.root = root.toAbsolutePath(); - this.indexPath = this.root.resolve("events.duckdb"); - this.partitionZone = partitionZone == null ? ZoneId.of("Asia/Shanghai") : partitionZone; - this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper; - this.writeParquetEnabled = writeParquetEnabled; - } - - @Override - public synchronized void appendAll(List records) throws IOException { - if (records == null || records.isEmpty()) { - return; - } - Map> byPartition = new LinkedHashMap<>(); - for (EventFileRecord record : records) { - Partition partition = partition(record); - byPartition.computeIfAbsent(partition, ignored -> new ArrayList<>()).add(record); - } - if (writeParquetEnabled) { - writePartitions(byPartition); - } - appendIndex(byPartition); - } - - @Override - public List query(EventFileQuery query) throws IOException { - // 有 VIN 且启用 Parquet 副本时，可直接扫分区文件；索引-only 模式走 DuckDB 主索引。 - if (query.vin() != null && writeParquetEnabled) { - return queryParquet(query); - } - // 无 VIN 的管理类查询走 DuckDB 索引，代价更高但不影响高频单车查询路径。 - ensureIndexInitialized(); - return queryIndex(query); - } - - @Override - public synchronized EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException { - if (rawArchiveUri == null || rawArchiveUri.isBlank()) { - return null; - } - ensureIndexInitialized(); - String sql = """ - SELECT event_id, protocol, event_type, vin, - event_time_ms, ingest_time_ms, - raw_archive_uri, metadata_json, payload_json - FROM event_records - WHERE raw_archive_uri = ? - ORDER BY event_type = 'RAW_ARCHIVE' DESC, ingest_time_ms DESC, event_id DESC - LIMIT 1 - """; - try (Connection connection = DriverManager.getConnection(indexJdbcUrl()); - PreparedStatement ps = connection.prepareStatement(sql)) { - ps.setString(1, rawArchiveUri); - try (ResultSet rs = ps.executeQuery()) { - return rs.next() ? record(rs) : null; - } - } catch (SQLException e) { - throw new IOException("query duckdb event index by raw archive uri failed", e); - } - } - - private List queryParquet(EventFileQuery query) throws IOException { - List parquetFiles = parquetFiles(query); - if (parquetFiles.isEmpty()) { - return List.of(); - } - String fileList = toDuckDbList(parquetFiles); - String order = query.order() == EventFileQuery.Order.DESC ? "DESC" : "ASC"; - List predicates = new ArrayList<>(); - if (query.vin() != null) { - predicates.add("vin = '" + escapeSql(query.vin()) + "'"); - } - if (query.eventType() != null) { - predicates.add("event_type = '" + escapeSql(query.eventType()) + "'"); - } - if (query.eventTimeFrom() != null) { - predicates.add("event_time_ms >= " + query.eventTimeFrom().toEpochMilli()); - } - if (query.eventTimeTo() != null) { - predicates.add("event_time_ms <= " + query.eventTimeTo().toEpochMilli()); - } - if (query.cursorEventTime() != null) { - predicates.add(cursorPredicateSql(query)); - } - // 这里的 SQL 只拼接经过 escape 的内部值；外部 rawArchiveUri 查询使用 PreparedStatement。 - String where = predicates.isEmpty() ? "" : "WHERE " + String.join(" AND ", predicates) + "\n"; - String sql = """ - SELECT event_id, protocol, event_type, vin, - event_time_ms, event_time, ingest_time_ms, ingest_time, - raw_archive_uri, metadata_json, payload_json - FROM read_parquet(%s) - %s - ORDER BY event_time_ms %s, ingest_time_ms %s, event_id %s - LIMIT %d - """.formatted(fileList, where, order, order, order, query.limit()); - try (Connection connection = DriverManager.getConnection("jdbc:duckdb:"); - Statement statement = connection.createStatement(); - ResultSet rs = statement.executeQuery(sql)) { - List result = new ArrayList<>(); - while (rs.next()) { - result.add(new EventFileRecord( - rs.getString("event_id"), - protocol(rs.getString("protocol")), - rs.getString("event_type"), - rs.getString("vin"), - Instant.ofEpochMilli(rs.getLong("event_time_ms")), - Instant.ofEpochMilli(rs.getLong("ingest_time_ms")), - rs.getString("raw_archive_uri"), - readMetadata(rs.getString("metadata_json")), - rs.getString("payload_json") - )); - } - return result; - } catch (SQLException e) { - throw new IOException("query parquet event store failed", e); - } - } - - private List queryIndex(EventFileQuery query) throws IOException { - String order = query.order() == EventFileQuery.Order.DESC ? "DESC" : "ASC"; - StringBuilder where = new StringBuilder(""" - WHERE protocol = ? - AND partition_date BETWEEN CAST(? AS DATE) AND CAST(? AS DATE) - """); - if (query.vin() != null) { - where.append(" AND vin = ?\n"); - } - if (query.eventType() != null) { - where.append(" AND event_type = ?\n"); - } - if (query.eventTimeFrom() != null) { - where.append(" AND event_time_ms >= ?\n"); - } - if (query.eventTimeTo() != null) { - where.append(" AND event_time_ms <= ?\n"); - } - if (query.cursorEventTime() != null) { - where.append(cursorPredicatePrepared(query.order())); - } - String sql = """ - SELECT event_id, protocol, event_type, vin, - event_time_ms, ingest_time_ms, - raw_archive_uri, metadata_json, payload_json - FROM event_records - %s - ORDER BY event_time_ms %s, ingest_time_ms %s, event_id %s - LIMIT ? - """.formatted(where, order, order, order); - try (Connection connection = DriverManager.getConnection(indexJdbcUrl()); - PreparedStatement ps = connection.prepareStatement(sql)) { - int index = 1; - ps.setString(index++, query.protocol().name()); - ps.setString(index++, query.dateFrom().toString()); - ps.setString(index++, query.dateTo().toString()); - if (query.vin() != null) { - ps.setString(index++, query.vin()); - } - if (query.eventType() != null) { - ps.setString(index++, query.eventType()); - } - if (query.eventTimeFrom() != null) { - ps.setLong(index++, query.eventTimeFrom().toEpochMilli()); - } - if (query.eventTimeTo() != null) { - ps.setLong(index++, query.eventTimeTo().toEpochMilli()); - } - if (query.cursorEventTime() != null) { - long eventTimeMs = query.cursorEventTime().toEpochMilli(); - long ingestTimeMs = query.cursorIngestTime().toEpochMilli(); - ps.setLong(index++, eventTimeMs); - ps.setLong(index++, eventTimeMs); - ps.setLong(index++, ingestTimeMs); - ps.setLong(index++, eventTimeMs); - ps.setLong(index++, ingestTimeMs); - ps.setString(index++, query.cursorEventId()); - } - ps.setInt(index, query.limit()); - try (ResultSet rs = ps.executeQuery()) { - List result = new ArrayList<>(); - while (rs.next()) { - result.add(record(rs)); - } - return result; - } - } catch (SQLException e) { - throw new IOException("query duckdb event index failed", e); - } - } - - private static String cursorPredicateSql(EventFileQuery query) { - long eventTimeMs = query.cursorEventTime().toEpochMilli(); - long ingestTimeMs = query.cursorIngestTime().toEpochMilli(); - String eventId = escapeSql(query.cursorEventId()); - String cmp = query.order() == EventFileQuery.Order.DESC ? "<" : ">"; - return """ - ( - event_time_ms %s %d - OR (event_time_ms = %d AND ingest_time_ms %s %d) - OR (event_time_ms = %d AND ingest_time_ms = %d AND event_id %s '%s') - ) - """.formatted(cmp, eventTimeMs, eventTimeMs, cmp, ingestTimeMs, - eventTimeMs, ingestTimeMs, cmp, eventId); - } - - private static String cursorPredicatePrepared(EventFileQuery.Order order) { - String cmp = order == EventFileQuery.Order.DESC ? "<" : ">"; - return """ - AND ( - event_time_ms %s ? - OR (event_time_ms = ? AND ingest_time_ms %s ?) - OR (event_time_ms = ? AND ingest_time_ms = ? AND event_id %s ?) - ) - """.formatted(cmp, cmp, cmp); - } - - private void writePartitions(Map> recordsByPartition) throws IOException { - if (recordsByPartition.isEmpty()) { - return; - } - List tempFiles = new ArrayList<>(); - - try (Connection connection = DriverManager.getConnection("jdbc:duckdb:"); - Statement statement = connection.createStatement()) { - statement.execute(""" - CREATE TEMPORARY TABLE event_records ( - partition_id INTEGER, - event_id VARCHAR, - protocol VARCHAR, - event_type VARCHAR, - vin VARCHAR, - event_time_ms BIGINT, - event_time VARCHAR, - ingest_time_ms BIGINT, - ingest_time VARCHAR, - raw_archive_uri VARCHAR, - metadata_json VARCHAR, - payload_json VARCHAR - ) - """); - try (PreparedStatement ps = connection.prepareStatement(""" - INSERT INTO event_records VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """)) { - int partitionId = 0; - for (List records : recordsByPartition.values()) { - for (EventFileRecord record : records) { - ps.setInt(1, partitionId); - ps.setString(2, record.eventId()); - ps.setString(3, record.protocol().name()); - ps.setString(4, record.eventType()); - ps.setString(5, record.vin()); - ps.setLong(6, record.eventTime().toEpochMilli()); - ps.setString(7, record.eventTime().toString()); - ps.setLong(8, record.ingestTime().toEpochMilli()); - ps.setString(9, record.ingestTime().toString()); - ps.setString(10, record.rawArchiveUri()); - ps.setString(11, objectMapper.writeValueAsString(record.metadata())); - ps.setString(12, record.payloadJson()); - ps.addBatch(); - } - partitionId++; - } - ps.executeBatch(); - } - int partitionId = 0; - for (Partition partition : recordsByPartition.keySet()) { - Path dir = partition.dir(root); - Files.createDirectories(dir); - String fragmentId = UUID.randomUUID().toString(); - Path partFile = dir.resolve("events-" + fragmentId + ".parquet"); - Path tempFile = dir.resolve("events-" + fragmentId + ".tmp.parquet"); - tempFiles.add(tempFile); - statement.execute(""" - COPY ( - SELECT event_id, protocol, event_type, vin, - event_time_ms, event_time, ingest_time_ms, ingest_time, - raw_archive_uri, metadata_json, payload_json - FROM event_records - WHERE partition_id = %d - ) TO '%s' (FORMAT parquet) - """.formatted(partitionId, escapeSql(tempFile))); - Files.move(tempFile, partFile, StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE); - partitionId++; - } - } catch (SQLException e) { - throw new IOException("write parquet event store failed", e); - } finally { - for (Path tempFile : tempFiles) { - Files.deleteIfExists(tempFile); - } - } - } - - private void appendIndex(Map> recordsByPartition) throws IOException { - ensureIndexInitialized(); - try (Connection connection = DriverManager.getConnection(indexJdbcUrl())) { - insertIndex(connection, recordsByPartition); - } catch (SQLException e) { - throw new IOException("write duckdb event index failed", e); - } - } - - private void ensureIndexInitialized() throws IOException { - if (indexInitialized) { - return; - } - try { - Files.createDirectories(root); - } catch (IOException e) { - throw new IOException("create event store root failed: " + root, e); - } - try (Connection connection = DriverManager.getConnection(indexJdbcUrl()); - Statement statement = connection.createStatement()) { - createIndexSchema(statement); - if (isIndexEmpty(statement)) { - // 新部署或删除 index 后，可从现有 parquet 自动回填索引，避免历史数据不可查。 - backfillIndexFromParquet(statement); - } - indexInitialized = true; - } catch (SQLException e) { - throw new IOException("initialize duckdb event index failed", e); - } - } - - private void createIndexSchema(Statement statement) throws SQLException { - statement.execute(""" - CREATE TABLE IF NOT EXISTS event_records ( - event_id VARCHAR PRIMARY KEY, - protocol VARCHAR, - event_type VARCHAR, - vin VARCHAR, - event_time_ms BIGINT, - ingest_time_ms BIGINT, - partition_date DATE, - raw_archive_uri VARCHAR, - metadata_json VARCHAR, - payload_json VARCHAR - ) - """); - statement.execute("CREATE INDEX IF NOT EXISTS event_records_protocol_date_time_idx ON event_records(protocol, partition_date, event_time_ms)"); - statement.execute("CREATE INDEX IF NOT EXISTS event_records_vin_date_time_idx ON event_records(vin, partition_date, event_time_ms)"); - statement.execute("CREATE INDEX IF NOT EXISTS event_records_vin_type_date_time_idx ON event_records(protocol, vin, event_type, partition_date, event_time_ms)"); - statement.execute("CREATE INDEX IF NOT EXISTS event_records_raw_archive_uri_idx ON event_records(raw_archive_uri)"); - } - - private boolean isIndexEmpty(Statement statement) throws SQLException { - try (ResultSet rs = statement.executeQuery("SELECT COUNT(*) FROM event_records")) { - return rs.next() && rs.getLong(1) == 0L; - } - } - - private void backfillIndexFromParquet(Statement statement) throws SQLException, IOException { - List files = allParquetFiles(); - if (files.isEmpty()) { - return; - } - String fileList = toDuckDbList(files); - statement.execute(""" - INSERT OR IGNORE INTO event_records - SELECT event_id, protocol, event_type, vin, - event_time_ms, ingest_time_ms, - CAST(date AS DATE) AS partition_date, - raw_archive_uri, metadata_json, payload_json - FROM read_parquet(%s, hive_partitioning = true) - """.formatted(fileList)); - } - - private void insertIndex(Connection connection, Map> recordsByPartition) - throws SQLException, IOException { - try (PreparedStatement ps = connection.prepareStatement(""" - INSERT OR REPLACE INTO event_records VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """)) { - for (Map.Entry> entry : recordsByPartition.entrySet()) { - Partition partition = entry.getKey(); - for (EventFileRecord record : entry.getValue()) { - ps.setString(1, record.eventId()); - ps.setString(2, record.protocol().name()); - ps.setString(3, record.eventType()); - ps.setString(4, record.vin()); - ps.setLong(5, record.eventTime().toEpochMilli()); - ps.setLong(6, record.ingestTime().toEpochMilli()); - ps.setString(7, partition.date().toString()); - ps.setString(8, record.rawArchiveUri()); - ps.setString(9, objectMapper.writeValueAsString(record.metadata())); - ps.setString(10, record.payloadJson()); - ps.addBatch(); - } - } - ps.executeBatch(); - } - } - - private List parquetFiles(EventFileQuery query) throws IOException { - List files = new ArrayList<>(); - LocalDate date = query.dateFrom(); - while (!date.isAfter(query.dateTo())) { - Path dateDir = partitionDir(query.protocol(), date); - List candidateDirs = query.vin() == null - ? List.of(dateDir) - : List.of( - // 新布局：高写入路径按协议/日期写一个批次文件，VIN 由 Parquet/索引字段过滤。 - dateDir.resolve("header=" + HEADER_VERSION), - // 兼容已落盘的旧布局。 - dateDir.resolve("bucket=" + bucketName(query.vin())), - dateDir.resolve("vehicle=" + storageName(query.vin()))); - for (Path dir : candidateDirs) { - if (Files.isDirectory(dir)) { - try (Stream stream = Files.walk(dir)) { - stream.filter(path -> path.getFileName().toString().endsWith(".parquet")) - .sorted() - .forEach(files::add); - } - } - } - date = date.plusDays(1); - } - return files; - } - - private List allParquetFiles() throws IOException { - if (!Files.isDirectory(root)) { - return List.of(); - } - try (Stream stream = Files.walk(root)) { - return stream.filter(path -> path.getFileName().toString().endsWith(".parquet")) - .sorted() - .toList(); - } - } - - private EventFileRecord record(ResultSet rs) throws SQLException, IOException { - return new EventFileRecord( - rs.getString("event_id"), - protocol(rs.getString("protocol")), - rs.getString("event_type"), - rs.getString("vin"), - Instant.ofEpochMilli(rs.getLong("event_time_ms")), - Instant.ofEpochMilli(rs.getLong("ingest_time_ms")), - rs.getString("raw_archive_uri"), - readMetadata(rs.getString("metadata_json")), - rs.getString("payload_json") - ); - } - - private Partition partition(EventFileRecord record) { - return new Partition( - record.protocol(), - LocalDate.ofInstant(record.eventTime(), partitionZone)); - } - - private Path partitionDir(ProtocolId protocol, LocalDate date) { - return root.resolve("protocol=" + protocol.name()).resolve("date=" + date); - } - - private static ProtocolId protocol(String value) { - if (value == null || value.isBlank()) { - return ProtocolId.UNKNOWN; - } - try { - return ProtocolId.valueOf(value); - } catch (IllegalArgumentException ex) { - return ProtocolId.UNKNOWN; - } - } - - private Map readMetadata(String json) throws IOException { - if (json == null || json.isBlank()) { - return Map.of(); - } - return objectMapper.readValue(json, STRING_MAP); - } - - private static String toDuckDbList(List files) { - return files.stream() - .map(path -> "'" + escapeSql(path) + "'") - .reduce((left, right) -> left + "," + right) - .map(paths -> "[" + paths + "]") - .orElse("[]"); - } - - private static String escapeSql(Path path) { - return escapeSql(path.toAbsolutePath().toString()); - } - - private static String escapeSql(String value) { - return value.replace("'", "''"); - } - - private String indexJdbcUrl() { - return "jdbc:duckdb:" + indexPath; - } - - private static String storageName(String vin) { - if (vin == null || vin.isBlank()) { - return "_unknown"; - } - // VIN 会进入文件路径，保守替换特殊字符，避免路径穿越和跨平台文件名问题。 - return vin.replaceAll("[^A-Za-z0-9._-]", "_"); - } - - private static String bucketName(String vin) { - String storageName = storageName(vin); - if ("_unknown".equals(storageName)) { - return "_unknown"; - } - return "%02d".formatted(Math.floorMod(storageName.hashCode(), VEHICLE_BUCKETS)); - } - - private record Partition(ProtocolId protocol, LocalDate date) { - private Path dir(Path root) { - return root.resolve("protocol=" + protocol.name()) - .resolve("date=" + date) - .resolve("header=" + HEADER_VERSION); - } - } -} diff --git a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileStoreSink.java b/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileStoreSink.java deleted file mode 100644 index 10233c7f..00000000 --- a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/EventFileStoreSink.java +++ /dev/null @@ -1,261 +0,0 @@ -package com.lingniu.ingest.eventfilestore; - -import com.fasterxml.jackson.databind.ObjectMapper; -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; -import java.util.ArrayList; -import java.util.LinkedHashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * EventBus 出口：把解析后的 {@link VehicleEvent} 明细写入 Parquet 文件库。 - * - *

{@link VehicleEvent.RawArchive} 在这里仅写可查询索引和 {@code archive://...} 引用，不写 - * 原始 bytes。本体必须已经由上游归档链路落成 .bin；否则 snapshot/frame 查询只能找到索引， - * 但回读原始包时会报缺失。 - * - *

32960 生产链路当前只把 RAW_ARCHIVE 作为历史查询事实源；REALTIME 和 LOCATION - * 是从 RAW 派生出来的轻量事件，不再重复写入 event-history，避免历史库里出现两套含义相近的数据。 - */ -public final class EventFileStoreSink implements EventSink, AutoCloseable { - - private static final Logger log = LoggerFactory.getLogger(EventFileStoreSink.class); - - private final EventFileStore store; - private final ObjectMapper objectMapper; - private final int batchSize; - private final List buffer = new ArrayList<>(); - private final ScheduledExecutorService flusher; - - public EventFileStoreSink(EventFileStore store, ObjectMapper objectMapper) { - this(store, objectMapper, 1); - } - - public EventFileStoreSink(EventFileStore store, ObjectMapper objectMapper, int batchSize) { - this(store, objectMapper, batchSize, 0); - } - - public EventFileStoreSink(EventFileStore store, - ObjectMapper objectMapper, - int batchSize, - long flushIntervalMillis) { - if (store == null) { - throw new IllegalArgumentException("store must not be null"); - } - this.store = store; - this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper; - this.batchSize = Math.max(1, batchSize); - this.flusher = flushIntervalMillis > 0 && this.batchSize > 1 - ? Executors.newSingleThreadScheduledExecutor(r -> { - Thread t = Thread.ofVirtual().name("event-file-store-flush-", 0).factory().newThread(r); - t.setDaemon(true); - return t; - }) - : null; - if (flusher != null) { - flusher.scheduleWithFixedDelay(this::flushQuietly, - flushIntervalMillis, flushIntervalMillis, TimeUnit.MILLISECONDS); - } - } - - @Override - public String name() { - return "event-file-store"; - } - - @Override - public boolean accepts(VehicleEvent event) { - // REALTIME/LOCATION 可以由 RAW 重放得到；历史库只保留 RAW_ARCHIVE 和少量非遥测事件索引。 - return event != null - && !(event instanceof VehicleEvent.Realtime) - && !(event instanceof VehicleEvent.Location); - } - - @Override - public CompletableFuture publish(VehicleEvent event) { - if (!accepts(event)) { - return CompletableFuture.completedFuture(null); - } - CompletableFuture future = new CompletableFuture<>(); - try { - EventFileRecord record = toRecord(event); - if (batchSize <= 1) { - store.append(record); - } else { - appendBuffered(record); - } - future.complete(null); - } catch (Exception e) { - future.completeExceptionally(e); - } - return future; - } - - @Override - public CompletableFuture publishBatch(List batch) { - if (batch == null || batch.isEmpty()) { - return CompletableFuture.completedFuture(null); - } - CompletableFuture future = new CompletableFuture<>(); - try { - List records = new ArrayList<>(); - for (VehicleEvent event : batch) { - if (accepts(event)) { - records.add(toRecord(event)); - } - } - if (!records.isEmpty()) { - store.appendAll(records); - } - future.complete(null); - } catch (Exception e) { - future.completeExceptionally(e); - } - return future; - } - - @Override - public void close() throws Exception { - if (flusher != null) { - flusher.shutdownNow(); - } - flush(); - } - - private void appendBuffered(EventFileRecord record) throws IOException { - List toFlush = null; - synchronized (buffer) { - buffer.add(record); - if (buffer.size() >= batchSize) { - // 拷贝后释放锁，避免慢 I/O 阻塞 EventBus 后续写入线程。 - toFlush = new ArrayList<>(buffer); - buffer.clear(); - } - } - if (toFlush != null) { - store.appendAll(toFlush); - } - } - - private void flush() throws IOException { - List toFlush; - synchronized (buffer) { - if (buffer.isEmpty()) { - return; - } - toFlush = new ArrayList<>(buffer); - buffer.clear(); - } - store.appendAll(toFlush); - } - - private void flushQuietly() { - try { - flush(); - } catch (Exception e) { - log.warn("event file store scheduled flush failed", e); - } - } - - private EventFileRecord toRecord(VehicleEvent event) throws IOException { - if (event instanceof VehicleEvent.RawArchive raw) { - return rawArchiveIndexRecord(raw); - } - TelemetrySnapshot snapshot = VehicleEventTelemetrySnapshotMapper.toSnapshot(event) - .orElseThrow(() -> new IllegalArgumentException("event cannot be converted to telemetry snapshot")); - return new EventFileRecord( - event.eventId(), - event.source(), - snapshot.eventType(), - event.vin(), - event.eventTime(), - event.ingestTime(), - snapshot.rawArchiveUri(), - event.metadata(), - objectMapper.writeValueAsString(snapshotPayload(snapshot)) - ); - } - - private EventFileRecord rawArchiveIndexRecord(VehicleEvent.RawArchive raw) throws IOException { - String rawArchiveKey = RawArchiveKeys.key(raw); - String rawArchiveUri = RawArchiveKeys.logicalUri(rawArchiveKey); - Map metadata = new LinkedHashMap<>(); - if (raw.metadata() != null) { - metadata.putAll(raw.metadata()); - } - // metadata 是 HTTP 查询和 raw 文件回查共用的关联字段，缺失时在这里补齐。 - metadata.putIfAbsent(RawArchiveKeys.META_EVENT_ID, raw.eventId()); - metadata.putIfAbsent(RawArchiveKeys.META_KEY, rawArchiveKey); - metadata.putIfAbsent(RawArchiveKeys.META_URI, rawArchiveUri); - return new EventFileRecord( - raw.eventId(), - raw.source(), - "RAW_ARCHIVE", - raw.vin(), - raw.eventTime(), - raw.ingestTime(), - rawArchiveUri, - metadata, - objectMapper.writeValueAsString(rawArchivePayload(raw, rawArchiveKey, rawArchiveUri)) - ); - } - - private static Map snapshotPayload(TelemetrySnapshot snapshot) { - Map payload = new LinkedHashMap<>(); - payload.put("eventId", snapshot.eventId()); - payload.put("vin", snapshot.vin()); - payload.put("protocol", snapshot.protocol().name()); - payload.put("eventType", snapshot.eventType()); - payload.put("eventTime", snapshot.eventTime().toString()); - payload.put("ingestTime", snapshot.ingestTime().toString()); - payload.put("rawArchiveUri", snapshot.rawArchiveUri()); - payload.put("metadata", snapshot.metadata()); - payload.put("fields", snapshot.fields()); - return payload; - } - - private Map rawArchivePayload(VehicleEvent.RawArchive raw, - String rawArchiveKey, - String rawArchiveUri) throws IOException { - Map payload = new LinkedHashMap<>(); - payload.put("eventId", raw.eventId()); - payload.put("vin", raw.vin()); - payload.put("protocol", raw.source().name()); - payload.put("eventType", "RAW_ARCHIVE"); - payload.put("eventTime", raw.eventTime().toString()); - payload.put("ingestTime", raw.ingestTime().toString()); - payload.put("rawArchiveUri", rawArchiveUri); - payload.put("rawArchiveKey", rawArchiveKey); - payload.put("command", hex(raw.command())); - payload.put("infoType", hex(raw.infoType())); - payload.put("rawSizeBytes", raw.rawBytes() == null ? 0 : raw.rawBytes().length); - putParsedJson(payload, raw.parsedJson()); - payload.put("metadata", raw.metadata()); - return payload; - } - - private void putParsedJson(Map payload, String parsedJson) throws IOException { - if (parsedJson == null || parsedJson.isBlank()) { - return; - } - payload.put("parsed", objectMapper.readTree(parsedJson)); - } - - private static String hex(int value) { - return "0x" + String.format("%04X", value); - } -} diff --git a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java b/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java deleted file mode 100644 index c92e5a56..00000000 --- a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java +++ /dev/null @@ -1,68 +0,0 @@ -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.EventFileStoreSink; -import org.springframework.boot.autoconfigure.AutoConfiguration; -import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.context.annotation.Bean; - -import java.nio.file.Path; -import java.time.ZoneId; - -/** - * EventFileStore 自动装配。 - * - *

开启 {@code lingniu.ingest.event-file-store.enabled=true} 后会创建两类 Bean： - *

    - *
  • {@link EventFileStore}：DuckDB sidecar index + Parquet 文件库，用于历史查询 - *
  • {@link EventFileStoreSink}：EventBus sink，把可接收事件转换为 {@code EventFileRecord} - *
- * - *

32960 当前生产路径里，event-file-store 主要保存 RAW_ARCHIVE 的可查询索引； - * 原始 bytes 本体由 sink-archive 写入 archive 根目录。 - */ -@AutoConfiguration -@EnableConfigurationProperties(EventFileStoreProperties.class) -@ConditionalOnProperty( - prefix = "lingniu.ingest.event-file-store", - name = "enabled", - havingValue = "true") -public class EventFileStoreAutoConfiguration { - - @Bean - @ConditionalOnMissingBean - public EventFileStore eventFileStore(EventFileStoreProperties properties, - ObjectProvider objectMapper) { - ObjectMapper mapper = mapper(objectMapper); - // zoneId 只影响文件分区日期；payload 中 eventTime/ingestTime 仍保持 Instant/UTC 表示。 - return new DuckDbParquetEventFileStore( - Path.of(properties.getPath()), - ZoneId.of(properties.getZoneId()), - mapper, - properties.isWriteParquetEnabled()); - } - - @Bean - @ConditionalOnMissingBean - public EventFileStoreSink eventFileStoreSink(EventFileStore store, - EventFileStoreProperties properties, - ObjectProvider objectMapper) { - return new EventFileStoreSink( - store, - mapper(objectMapper), - properties.getBatchSize(), - properties.getFlushIntervalMillis()); - } - - private static ObjectMapper mapper(ObjectProvider provider) { - ObjectMapper base = provider.getIfAvailable(ObjectMapper::new); - // 使用 copy，避免全局 ObjectMapper 被本模块注册 JavaTimeModule 时产生隐式副作用。 - return base.copy().registerModule(new JavaTimeModule()); - } -} diff --git a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java b/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java deleted file mode 100644 index 4db78bca..00000000 --- a/modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java +++ /dev/null @@ -1,97 +0,0 @@ -package com.lingniu.ingest.eventfilestore.config; - -import org.springframework.boot.context.properties.ConfigurationProperties; - -@ConfigurationProperties(prefix = "lingniu.ingest.event-file-store") -public class EventFileStoreProperties { - - /** - * 默认关闭，避免仅引入依赖就开始写明细文件。 - */ - private boolean enabled = false; - - /** - * 历史库根路径。 - * - *

{@code events.duckdb} 是可查询主索引；开启 Parquet 副本时，还会在该根目录下按 - * {@code protocol/date/header} 保存追加文件。 - */ - private String path = "./event-store"; - - /** - * 按事件时间分区时使用的业务时区。 - * - *

该配置只决定“某条事件落入哪一天目录”，不改变接口响应里的 UTC 时间字符串。 - */ - private String zoneId = "Asia/Shanghai"; - - /** - * Sink 缓冲多少条事件后批量写一个 Parquet part 文件。 - * - *

值越大写入吞吐越好，但异常退出时内存缓冲里尚未 flush 的事件越多。 - */ - private int batchSize = 500; - - /** - * 未达到 batchSize 时的最长刷盘间隔。 - * - *

用于低流量车辆：即使没有攒满 batch，也会定期把缓冲写入文件，降低查询延迟。 - */ - private long flushIntervalMillis = 1000; - - /** - * 是否同步写 Parquet 副本。 - * - *

高写入生产链路可以关闭，只保留 DuckDB JSON 主索引和 raw archive 引用；需要离线副本或 - * 索引重建能力时再开启。 - */ - private boolean writeParquetEnabled = true; - - public boolean isEnabled() { - return enabled; - } - - public void setEnabled(boolean enabled) { - this.enabled = enabled; - } - - public String getPath() { - return path; - } - - public void setPath(String path) { - this.path = path; - } - - public String getZoneId() { - return zoneId; - } - - public void setZoneId(String zoneId) { - this.zoneId = zoneId; - } - - public int getBatchSize() { - return batchSize; - } - - public void setBatchSize(int batchSize) { - this.batchSize = batchSize; - } - - public long getFlushIntervalMillis() { - return flushIntervalMillis; - } - - public void setFlushIntervalMillis(long flushIntervalMillis) { - this.flushIntervalMillis = flushIntervalMillis; - } - - public boolean isWriteParquetEnabled() { - return writeParquetEnabled; - } - - public void setWriteParquetEnabled(boolean writeParquetEnabled) { - this.writeParquetEnabled = writeParquetEnabled; - } -} diff --git a/modules/sinks/event-file-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/modules/sinks/event-file-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports deleted file mode 100644 index 4a2b4534..00000000 --- a/modules/sinks/event-file-store/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ /dev/null @@ -1 +0,0 @@ -com.lingniu.ingest.eventfilestore.config.EventFileStoreAutoConfiguration diff --git a/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStoreTest.java b/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStoreTest.java deleted file mode 100644 index 567d169c..00000000 --- a/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStoreTest.java +++ /dev/null @@ -1,319 +0,0 @@ -package com.lingniu.ingest.eventfilestore; - -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; - -import java.nio.file.Files; -import java.nio.file.Path; -import java.time.Instant; -import java.time.LocalDate; -import java.time.ZoneId; -import java.util.List; -import java.util.Map; -import java.util.stream.IntStream; - -import static org.assertj.core.api.Assertions.assertThat; - -class DuckDbParquetEventFileStoreTest { - - @TempDir - Path tempDir; - - @Test - void appendsRecordsToParquetAndQueriesByTimeAscendingAndDescending() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - EventFileRecord first = record("e1", Instant.parse("2026-06-21T16:00:02Z"), "粤A00001"); - EventFileRecord second = record("e2", Instant.parse("2026-06-21T16:00:01Z"), "粤A00002"); - - store.appendAll(List.of(first, second)); - - Path dateDir = tempDir.resolve("protocol=GB32960").resolve("date=2026-06-22"); - assertThat(parquetFiles(dateDir)).hasSize(1); - - EventFileQuery ascending = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.ASC, - 100); - assertThat(store.query(ascending)) - .extracting(EventFileRecord::eventId) - .containsExactly("e2", "e1"); - - EventFileQuery descending = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.DESC, - 100); - assertThat(store.query(descending)) - .extracting(EventFileRecord::eventId) - .containsExactly("e1", "e2"); - } - - @Test - void queriesAcrossMultipleDaysInTimeOrder() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - store.append(record("d1", Instant.parse("2026-06-20T16:00:00Z"), "粤A00001")); - store.append(record("d2", Instant.parse("2026-06-21T16:00:00Z"), "粤A00001")); - - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-21"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.ASC, - 100); - - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("d1", "d2"); - } - - @Test - void filtersByVinBeforeApplyingLimit() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - store.append(record("other-1", Instant.parse("2026-06-21T16:00:03Z"), "VIN-OTHER")); - store.append(record("target-1", Instant.parse("2026-06-21T16:00:02Z"), "VIN-TARGET")); - store.append(record("other-2", Instant.parse("2026-06-21T16:00:01Z"), "VIN-OTHER")); - store.append(record("target-2", Instant.parse("2026-06-21T16:00:00Z"), "VIN-TARGET")); - - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.DESC, - 1, - "VIN-TARGET"); - - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("target-1"); - } - - @Test - void filtersByVinAndEventTypeBeforeApplyingLimit() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - store.append(rawRecord("raw-other", Instant.parse("2026-06-21T16:00:04Z"), "VIN-OTHER")); - store.append(record("business-target", Instant.parse("2026-06-21T16:00:03Z"), "VIN-TARGET")); - store.append(rawRecord("raw-target", Instant.parse("2026-06-21T16:00:02Z"), "VIN-TARGET")); - - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.DESC, - 1, - "VIN-TARGET", - "RAW_ARCHIVE"); - - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("raw-target"); - } - - @Test - void filtersByExactEventTimeRangeBeforeApplyingLimit() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - store.append(rawRecord("before", Instant.parse("2026-06-21T16:00:00Z"), "VIN-TIME")); - store.append(rawRecord("inside", Instant.parse("2026-06-21T16:00:02Z"), "VIN-TIME")); - store.append(rawRecord("after", Instant.parse("2026-06-21T16:00:04Z"), "VIN-TIME")); - - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - Instant.parse("2026-06-21T16:00:01Z"), - Instant.parse("2026-06-21T16:00:03Z"), - EventFileQuery.Order.ASC, - 10, - "VIN-TIME", - "RAW_ARCHIVE"); - - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("inside"); - } - - @Test - void appendsSameVehicleDayHeaderAsAppendOnlyParquetFragments() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - store.append(record("first", Instant.parse("2026-06-21T16:00:03Z"), "VIN-SAME")); - store.append(record("second", Instant.parse("2026-06-21T16:00:04Z"), "VIN-SAME")); - - Path dateDir = tempDir.resolve("protocol=GB32960").resolve("date=2026-06-22"); - assertThat(parquetFiles(dateDir)).hasSize(2); - - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.ASC, - 10, - "VIN-SAME"); - - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("first", "second"); - } - - @Test - void writesManyVehiclesIntoBoundedDailyLayoutInsteadOfVehicleDirectories() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - List records = IntStream.range(0, 80) - .mapToObj(index -> rawRecord( - "raw-" + index, - Instant.parse("2026-06-21T16:00:00Z").plusMillis(index), - "VIN-" + index)) - .toList(); - - store.appendAll(records); - - Path dateDir = tempDir.resolve("protocol=GB32960").resolve("date=2026-06-22"); - try (var files = Files.list(dateDir)) { - List partitionDirs = files - .filter(Files::isDirectory) - .map(path -> path.getFileName().toString()) - .sorted() - .toList(); - assertThat(partitionDirs).noneMatch(name -> name.startsWith("vehicle=")); - assertThat(partitionDirs).hasSizeLessThan(records.size()); - } - } - - @Test - void writesOneDailyParquetFragmentForOneBatchAcrossManyVehicles() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - List records = IntStream.range(0, 80) - .mapToObj(index -> rawRecord( - "daily-raw-" + index, - Instant.parse("2026-06-21T16:00:00Z").plusMillis(index), - "VIN-DAILY-" + index)) - .toList(); - - store.appendAll(records); - - Path dateDir = tempDir.resolve("protocol=GB32960").resolve("date=2026-06-22"); - assertThat(parquetFiles(dateDir)).hasSize(1); - } - - @Test - void queriesUseSidecarIndexInsteadOfScanningParquetPartsEveryTime() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - store.append(record("indexed", Instant.parse("2026-06-21T16:00:03Z"), "VIN-INDEXED")); - - Path dateDir = tempDir.resolve("protocol=GB32960").resolve("date=2026-06-22"); - for (Path file : parquetFiles(dateDir)) { - Files.delete(file); - } - - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.DESC, - 10); - - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("indexed"); - } - - @Test - void vinQueriesReadVehicleDayParquetWithoutGlobalDuckDbIndex() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - store.append(rawRecord("raw-target", Instant.parse("2026-06-21T16:00:03Z"), "VIN-DIRECT")); - Files.deleteIfExists(tempDir.resolve("events.duckdb")); - - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.DESC, - 10, - "VIN-DIRECT", - "RAW_ARCHIVE"); - - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("raw-target"); - } - - @Test - void findsRecordByRawArchiveUriThroughSidecarIndex() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai")); - EventFileRecord record = record("raw-001", Instant.parse("2026-06-21T16:00:03Z"), "VIN-RAW"); - store.append(record); - - EventFileRecord found = store.findByRawArchiveUri(record.rawArchiveUri()); - - assertThat(found).isNotNull(); - assertThat(found.eventId()).isEqualTo("raw-001"); - assertThat(found.rawArchiveUri()).isEqualTo(record.rawArchiveUri()); - } - - @Test - void canQueryVinRecordsFromIndexWhenParquetWritingIsDisabled() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore( - tempDir, - ZoneId.of("Asia/Shanghai"), - new ObjectMapper(), - false); - EventFileRecord record = rawRecord("index-only", Instant.parse("2026-06-21T16:00:03Z"), "VIN-INDEX-ONLY"); - - store.append(record); - - assertThat(parquetFiles(tempDir)).isEmpty(); - EventFileQuery query = new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.DESC, - 10, - "VIN-INDEX-ONLY", - "RAW_ARCHIVE"); - assertThat(store.query(query)) - .extracting(EventFileRecord::eventId) - .containsExactly("index-only"); - } - - private static List parquetFiles(Path root) throws Exception { - try (var files = Files.walk(root)) { - return files.filter(p -> p.getFileName().toString().endsWith(".parquet")) - .sorted() - .toList(); - } - } - - private static EventFileRecord record(String eventId, Instant eventTime, String vin) { - return new EventFileRecord( - eventId, - ProtocolId.GB32960, - "REALTIME", - vin, - eventTime, - eventTime.plusMillis(200), - "file:///archive/" + eventId + ".bin", - Map.of("source", "test"), - """ - {"gb32960":{"vehicle":{"speedKmh":42.1},"fuelCell":{"hydrogenRemainingKg":8.3}}} - """.strip()); - } - - private static EventFileRecord rawRecord(String eventId, Instant eventTime, String vin) { - return new EventFileRecord( - eventId, - ProtocolId.GB32960, - "RAW_ARCHIVE", - vin, - eventTime, - eventTime.plusMillis(200), - "file:///archive/" + eventId + ".bin", - Map.of("source", "test"), - "{}"); - } -} diff --git a/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/EventFileStoreSinkTest.java b/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/EventFileStoreSinkTest.java deleted file mode 100644 index cc52f074..00000000 --- a/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/EventFileStoreSinkTest.java +++ /dev/null @@ -1,157 +0,0 @@ -package com.lingniu.ingest.eventfilestore; - -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.lingniu.ingest.api.ProtocolId; -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; - -import java.nio.file.Path; -import java.time.Instant; -import java.time.LocalDate; -import java.time.ZoneId; -import java.util.Map; - -import static org.assertj.core.api.Assertions.assertThat; - -class EventFileStoreSinkTest { - - @TempDir - Path tempDir; - - private final ObjectMapper objectMapper = new ObjectMapper(); - - @Test - void eventSinkSkipsRealtimeAndLocationEvents() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), objectMapper); - EventFileStoreSink sink = new EventFileStoreSink(store, objectMapper); - - sink.publish(realtimeEvent("evt-1")).join(); - sink.publish(locationEvent("loc-1")).join(); - - var records = store.query(new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.ASC, - 10)); - - assertThat(records).isEmpty(); - } - - @Test - void batchedSinkFlushesRecordsWhenClosed() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), objectMapper); - EventFileStoreSink sink = new EventFileStoreSink(store, objectMapper, 10); - - sink.publish(rawEvent("raw-batch-1")).join(); - assertThat(store.query(new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.ASC, - 10))).isEmpty(); - - sink.close(); - - assertThat(store.query(new EventFileQuery( - ProtocolId.GB32960, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.ASC, - 10))) - .extracting(EventFileRecord::eventId) - .containsExactly("raw-batch-1"); - } - - @Test - void eventSinkPersistsRawArchiveAsQueryableIndexWithoutEmbeddingRawBytes() throws Exception { - EventFileStore store = new DuckDbParquetEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), objectMapper); - EventFileStoreSink sink = new EventFileStoreSink(store, objectMapper); - VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive( - "raw-1", - "LTEST000000000808", - ProtocolId.JT808, - Instant.parse("2026-06-21T16:00:03Z"), - Instant.parse("2026-06-21T16:00:04Z"), - "trace-raw-1", - Map.of("phone", "013800138000", "identityResolved", "true"), - 0x0200, - 0, - new byte[]{0x7e, 0x02, 0x00, 0x7e}); - - sink.publish(raw).join(); - - var records = store.query(new EventFileQuery( - ProtocolId.JT808, - LocalDate.parse("2026-06-22"), - LocalDate.parse("2026-06-22"), - EventFileQuery.Order.ASC, - 10)); - - assertThat(records).hasSize(1); - EventFileRecord record = records.getFirst(); - String rawArchiveKey = RawArchiveKeys.key(raw); - assertThat(record.eventId()).isEqualTo("raw-1"); - assertThat(record.eventType()).isEqualTo("RAW_ARCHIVE"); - assertThat(record.rawArchiveUri()).isEqualTo(RawArchiveKeys.logicalUri(rawArchiveKey)); - assertThat(record.metadata()) - .containsEntry(RawArchiveKeys.META_KEY, rawArchiveKey) - .containsEntry(RawArchiveKeys.META_URI, RawArchiveKeys.logicalUri(rawArchiveKey)); - JsonNode payload = objectMapper.readTree(record.payloadJson()); - assertThat(payload.get("rawSizeBytes").asLong()).isEqualTo(4); - assertThat(payload.get("command").asText()).isEqualTo("0x0200"); - assertThat(payload.toString()).doesNotContain("7e02007e"); - } - - private static VehicleEvent.Realtime realtimeEvent(String eventId) { - return new VehicleEvent.Realtime( - eventId, - "LTEST000000000001", - ProtocolId.GB32960, - Instant.parse("2026-06-21T16:00:00Z"), - Instant.parse("2026-06-21T16:00:01Z"), - "trace-1", - Map.of("rawArchiveUri", "file:///archive/" + eventId + ".bin"), - new RealtimePayload( - 42.1, 1234.5, 87.0, null, null, - null, null, null, 8.3, null, null, - RealtimePayload.VehicleState.STARTED, - null, null, null, null, null, - 113.12, 23.45, null, null, null, null - )); - } - - private static VehicleEvent.Location locationEvent(String eventId) { - return new VehicleEvent.Location( - eventId, - "LTEST000000000001", - ProtocolId.GB32960, - Instant.parse("2026-06-21T16:00:00Z"), - Instant.parse("2026-06-21T16:00:01Z"), - "trace-location", - Map.of("rawArchiveUri", "file:///archive/" + eventId + ".bin"), - new LocationPayload(113.12, 23.45, 0, 42.1, 90, 0, 0)); - } - - private static VehicleEvent.RawArchive rawEvent(String eventId) { - return new VehicleEvent.RawArchive( - eventId, - "LTEST000000000001", - ProtocolId.GB32960, - Instant.parse("2026-06-21T16:00:00Z"), - Instant.parse("2026-06-21T16:00:01Z"), - "trace-raw", - Map.of(), - 0x0002, - 0, - new byte[]{0x23, 0x23}); - } -} diff --git a/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java b/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java deleted file mode 100644 index 66f58f2e..00000000 --- a/modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java +++ /dev/null @@ -1,44 +0,0 @@ -package com.lingniu.ingest.eventfilestore.config; - -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; -import org.springframework.boot.test.context.runner.ApplicationContextRunner; - -import java.nio.file.Path; - -import static org.assertj.core.api.Assertions.assertThat; - -class EventFileStoreAutoConfigurationTest { - - private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() - .withConfiguration(AutoConfigurations.of(EventFileStoreAutoConfiguration.class)); - - @Test - void createsStoreAndSinkWhenEnabled() { - contextRunner - .withPropertyValues( - "lingniu.ingest.event-file-store.enabled=true", - "lingniu.ingest.event-file-store.path=target/test-event-store", - "lingniu.ingest.event-file-store.write-parquet-enabled=false") - .run(context -> { - assertThat(context).hasSingleBean(EventFileStore.class); - assertThat(context).hasSingleBean(EventFileStoreSink.class); - assertThat(context.getBean(EventFileStoreProperties.class).getPath()) - .isEqualTo(Path.of("target/test-event-store").toString()); - assertThat(context.getBean(EventFileStoreProperties.class).isWriteParquetEnabled()) - .isFalse(); - }); - } - - @Test - void backsOffWhenDisabled() { - contextRunner - .withPropertyValues("lingniu.ingest.event-file-store.enabled=false") - .run(context -> { - assertThat(context).doesNotHaveBean(EventFileStore.class); - assertThat(context).doesNotHaveBean(EventFileStoreSink.class); - }); - } -} diff --git a/pom.xml b/pom.xml index 4772b126..400708ae 100644 --- a/pom.xml +++ b/pom.xml @@ -106,21 +106,6 @@ - - optional-event-file-store - - modules/sinks/event-file-store - - - - - com.lingniu.ingest - event-file-store - ${project.version} - - - - legacy-xinda @@ -164,7 +149,6 @@ 1.42.1 1.2.5 1.12 - 1.1.3 3.8.4 2.8.17 1.18.34 @@ -374,11 +358,6 @@ mqtt-client ${fusesource-mqtt.version} - - org.duckdb - duckdb_jdbc - ${duckdb.version} - com.taosdata.jdbc taos-jdbcdriver