You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "XuQianJin-Stars (via GitHub)" <gi...@apache.org> on 2026/02/05 08:29:47 UTC
[PR] [lake/lance] Add NestedRow type support for Lance [fluss]
XuQianJin-Stars opened a new pull request, #2578:
URL: https://github.com/apache/fluss/pull/2578
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #2404
<!-- What is the purpose of the change -->
This PR adds NestedRow (Struct) type support for Lance lake storage, extending the existing Array type support implementation.
### Brief change log
<!-- Please describe the changes made in this pull request and explain how they address the issue -->
- **LanceArrowUtils.java**:
- Extended `toArrowField()` to handle `RowType` by recursively creating child fields for nested struct types
- Extended `toArrowType()` to map Fluss `RowType` to Arrow `Struct.INSTANCE`
- **ArrowDataConverter.java**:
- Added `copyStructVectorData()` method to recursively copy data from shaded `StructVector` to non-shaded `StructVector`
- Updated `copyVectorData()` to detect and delegate to struct-specific copy logic
- **ShadedArrowBatchWriter.java**:
- Extended `initFieldVector()` to properly allocate and initialize `StructVector` and its child vectors
- **FlinkLanceTieringTestBase.java**:
- Added `createLogTableWithNestedRowType()` helper method for creating tables with nested Row columns
- Added `createLogTableWithArrayOfRowType()` helper method for creating tables with Array<Row> columns
### Tests
<!-- List UT and IT cases to verify this change -->
**Unit Tests (LanceArrowUtilsTest.java)**:
- `testToArrowSchemaWithNestedRowType`: Verifies simple nested Row type conversion to Arrow Struct
- `testToArrowSchemaWithDeeplyNestedRowType`: Verifies deeply nested Row type conversion
- `testToArrowSchemaWithArrayOfRowType`: Verifies Array<Row> type conversion
- `testToArrowSchemaWithRowContainingArray`: Verifies Row containing Array field
**Unit Tests (LanceTieringTest.java)**:
- `testTieringWriteTableWithNestedRowType`: Verifies writing and reading tables with nested Row type
**Integration Tests (LanceTieringITCase.java)**:
- `testTieringWithNestedRowType`: End-to-end test for tiering with nested Row type
- `testTieringWithArrayOfRowType`: End-to-end test for tiering with Array<Row> type
### API and Format
<!-- Does this change affect API or storage format -->
No API changes. This change extends the internal Lance lake storage format to support Struct types, which is backward compatible.
### Documentation
<!-- Does this change introduce a new feature -->
No documentation changes needed. This is an internal enhancement to support additional data types in Lance lake storage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add NestedRow type support for Lance [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #2578:
URL: https://github.com/apache/fluss/pull/2578#discussion_r2956282020
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringITCase.java:
##########
@@ -140,6 +142,112 @@ void testTiering() throws Exception {
jobClient.cancel().get();
}
+ @Test
+ void testTieringWithNestedRowType() throws Exception {
Review Comment:
Nit: should we test row of row as well?
##########
fluss-lake/fluss-lake-lance/src/main/java/org/apache/fluss/lake/lance/utils/ArrowDataConverter.java:
##########
@@ -94,6 +95,17 @@ private static void copyVectorData(
}
}
+ if (shadedVector
Review Comment:
I wonder if we should be more defensive in this class. Currently if only one of the vector class is struct, this will be skipped and use non-struct copy logic, failure will appear much later. (Appreciate that this is the convention of the class i.e. listvector, consider this as nit)
##########
website/docs/_configs/_partial_config.mdx:
##########
@@ -284,7 +288,7 @@
| `table.auto-partition.enabled` | `false` | Boolean | Whether enable auto partition for the table. Disable by default. When auto partition is enabled, the partitions of the table will be created automatically. |
| `table.auto-partition.key` | `none` | String | This configuration defines the time-based partition key to be used for auto-partitioning when a table is partitioned with multiple keys. Auto-partitioning utilizes a time-based partition key to handle partitions automatically, including creating new ones and removing outdated ones, by comparing the time value of the partition with the current system time. In the case of a table using multiple partition keys (such as a composite partitioning strategy), this feature determines which key should serve as the primary time dimension for making auto-partitioning decisions.And If the table has only one partition key, this config is not necessary. Otherwise, it must be specified. |
| `table.auto-partition.time-unit` | `DAY` | AutoPartitionTimeUnit | The time granularity for auto created partitions. The default value is `DAY`. Valid values are `HOUR`, `DAY`, `MONTH`, `QUARTER`, `YEAR`. If the value is `HOUR`, the partition format for auto created is yyyyMMddHH. If the value is `DAY`, the partition format for auto created is yyyyMMdd. If the value is `MONTH`, the partition format for auto created is yyyyMM. If the value is `QUARTER`, the partition format for auto created is yyyyQ. If the value is `YEAR`, the partition format for auto created is yyyy. |
-| `table.auto-partition.time-zone` | `Europe/Paris` | String | The time zone for auto partitions, which is by default the same as the system time zone. |
+| `table.auto-partition.time-zone` | `Asia/Shanghai` | String | The time zone for auto partitions, which is by default the same as the system time zone. |
Review Comment:
Ditto
##########
website/docs/_configs/_partial_config.mdx:
##########
@@ -238,9 +238,13 @@
| Key | Default | Type | Description |
| :--- | :--- | :--- | :--- |
-| `remote.data.dir` | `none` | String | The directory used for storing the kv snapshot data files and remote log for log tiered storage in a Fluss supported filesystem. |
+| `remote.data.dir` | `none` | String | The directory used for storing the kv snapshot data files and remote log for log tiered storage in a Fluss supported filesystem. When upgrading to `remote.data.dirs`, please ensure this value is placed as the first entry in the new configuration.For new clusters, it is recommended to use `remote.data.dirs` instead. If `remote.data.dirs` is configured, this value will be ignored. |
+| `remote.data.dirs` | `[]` | ArrayList | A comma-separated list of directories in Fluss supported filesystems for storing the kv snapshot data files and remote log files of tables/partitions. If configured, when a new table or a new partition is created, one of the directories from this list will be selected according to the strategy specified by `remote.data.dirs.strategy` (`ROUND_ROBIN` by default). If not configured, the system uses `remote.data.dir` as the sole remote data directory for all data. |
Review Comment:
Are these part of the lance row support change?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add NestedRow type support for Lance [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #2578:
URL: https://github.com/apache/fluss/pull/2578#discussion_r2972160077
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java:
##########
@@ -355,4 +356,170 @@ private Schema createTable(LanceConfig config, Map<String, String> customPropert
return schema;
}
+
+ @ParameterizedTest
+ @MethodSource("tieringWriteArgs")
+ void testTieringWriteTableWithNestedRowType(boolean isPartitioned) throws Exception {
+ int bucketNum = 3;
+ TablePath tablePath = TablePath.of("lance", "nestedRowTable");
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("lance.batch_size", "256");
+ LanceConfig config =
+ LanceConfig.from(
+ configuration.toMap(),
+ customProperties,
+ tablePath.getDatabaseName(),
+ tablePath.getTableName());
+ Schema schema = createNestedRowTable(config);
+
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(bucketNum)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .customProperties(customProperties)
+ .build();
+ TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
+ List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
+ SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
+ lanceLakeTieringFactory.getWriteResultSerializer();
+ SimpleVersionedSerializer<LanceCommittable> committableSerializer =
+ lanceLakeTieringFactory.getCommittableSerializer();
+
+ Map<Tuple2<String, Integer>, List<LogRecord>> recordsByBucket = new HashMap<>();
+ Map<Long, String> partitionIdAndName =
+ isPartitioned
+ ? new HashMap<Long, String>() {
+ {
+ put(1L, "p1");
+ put(2L, "p2");
+ put(3L, "p3");
+ }
+ }
+ : Collections.singletonMap(null, null);
+
+ // First, write data with nested row types
+ for (int bucket = 0; bucket < bucketNum; bucket++) {
+ for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
+ String partition = entry.getValue();
+ try (LakeWriter<LanceWriteResult> lakeWriter =
+ createLakeWriter(tablePath, bucket, partition, tableInfo)) {
+ Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
+ Tuple2<List<LogRecord>, List<LogRecord>> writeAndExpectRecords =
+ genNestedRowLogRecords(partition, bucket, 10);
+ List<LogRecord> writtenRecords = writeAndExpectRecords.f0;
+ List<LogRecord> expectRecords = writeAndExpectRecords.f1;
+ recordsByBucket.put(partitionBucket, expectRecords);
+ for (LogRecord logRecord : writtenRecords) {
+ lakeWriter.write(logRecord);
+ }
+ // serialize/deserialize writeResult
+ LanceWriteResult lanceWriteResult = lakeWriter.complete();
+ byte[] serialized = writeResultSerializer.serialize(lanceWriteResult);
+ lanceWriteResults.add(
+ writeResultSerializer.deserialize(
+ writeResultSerializer.getVersion(), serialized));
+ }
+ }
+ }
+
+ // Second, commit data
+ try (LakeCommitter<LanceWriteResult, LanceCommittable> lakeCommitter =
+ createLakeCommitter(tablePath, tableInfo)) {
+ // serialize/deserialize committable
+ LanceCommittable lanceCommittable = lakeCommitter.toCommittable(lanceWriteResults);
+ byte[] serialized = committableSerializer.serialize(lanceCommittable);
+ lanceCommittable =
+ committableSerializer.deserialize(
+ committableSerializer.getVersion(), serialized);
+ Map<String, String> snapshotProperties =
+ Collections.singletonMap(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets");
+ LakeCommitResult commitResult =
+ lakeCommitter.commit(lanceCommittable, snapshotProperties);
+ // lance dataset version starts from 1
+ assertThat(commitResult.getCommittedSnapshotId()).isEqualTo(2);
+ }
+
+ try (Dataset dataset =
+ Dataset.open(
+ new RootAllocator(),
+ config.getDatasetUri(),
+ LanceConfig.genReadOptionFromConfig(config))) {
+ ArrowReader reader = dataset.newScan().scanBatches();
+ VectorSchemaRoot readerRoot = reader.getVectorSchemaRoot();
+
+ // Verify data can be read back
+ for (int bucket = 0; bucket < 3; bucket++) {
+ for (String partition : partitionIdAndName.values()) {
+ reader.loadNextBatch();
+ Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
+ List<LogRecord> expectRecords = recordsByBucket.get(partitionBucket);
+ verifyNestedRowRecords(readerRoot, expectRecords, bucket, isPartitioned);
+ }
+ }
+ assertThat(reader.loadNextBatch()).isFalse();
+ }
+ }
+
+ private Schema createNestedRowTable(LanceConfig config) {
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column(
+ "address",
+ DataTypes.ROW(
+ DataTypes.FIELD("city", DataTypes.STRING()),
+ DataTypes.FIELD("zip", DataTypes.INT())));
+ Schema schema = schemaBuilder.build();
+ WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
+ LanceDatasetAdapter.createDataset(
+ config.getDatasetUri(), LanceArrowUtils.toArrowSchema(schema.getRowType()), params);
+
+ return schema;
+ }
+
+ private Tuple2<List<LogRecord>, List<LogRecord>> genNestedRowLogRecords(
+ @Nullable String partition, int bucket, int numRecords) {
Review Comment:
`partition` is unused, remove?
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java:
##########
@@ -355,4 +356,170 @@ private Schema createTable(LanceConfig config, Map<String, String> customPropert
return schema;
}
+
+ @ParameterizedTest
+ @MethodSource("tieringWriteArgs")
+ void testTieringWriteTableWithNestedRowType(boolean isPartitioned) throws Exception {
+ int bucketNum = 3;
+ TablePath tablePath = TablePath.of("lance", "nestedRowTable");
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("lance.batch_size", "256");
+ LanceConfig config =
+ LanceConfig.from(
+ configuration.toMap(),
+ customProperties,
+ tablePath.getDatabaseName(),
+ tablePath.getTableName());
+ Schema schema = createNestedRowTable(config);
+
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(bucketNum)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .customProperties(customProperties)
+ .build();
+ TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
+ List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
+ SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
+ lanceLakeTieringFactory.getWriteResultSerializer();
+ SimpleVersionedSerializer<LanceCommittable> committableSerializer =
+ lanceLakeTieringFactory.getCommittableSerializer();
+
+ Map<Tuple2<String, Integer>, List<LogRecord>> recordsByBucket = new HashMap<>();
+ Map<Long, String> partitionIdAndName =
+ isPartitioned
+ ? new HashMap<Long, String>() {
+ {
+ put(1L, "p1");
+ put(2L, "p2");
+ put(3L, "p3");
+ }
+ }
+ : Collections.singletonMap(null, null);
+
+ // First, write data with nested row types
+ for (int bucket = 0; bucket < bucketNum; bucket++) {
+ for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
+ String partition = entry.getValue();
+ try (LakeWriter<LanceWriteResult> lakeWriter =
+ createLakeWriter(tablePath, bucket, partition, tableInfo)) {
+ Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
+ Tuple2<List<LogRecord>, List<LogRecord>> writeAndExpectRecords =
+ genNestedRowLogRecords(partition, bucket, 10);
+ List<LogRecord> writtenRecords = writeAndExpectRecords.f0;
+ List<LogRecord> expectRecords = writeAndExpectRecords.f1;
+ recordsByBucket.put(partitionBucket, expectRecords);
+ for (LogRecord logRecord : writtenRecords) {
+ lakeWriter.write(logRecord);
+ }
+ // serialize/deserialize writeResult
+ LanceWriteResult lanceWriteResult = lakeWriter.complete();
+ byte[] serialized = writeResultSerializer.serialize(lanceWriteResult);
+ lanceWriteResults.add(
+ writeResultSerializer.deserialize(
+ writeResultSerializer.getVersion(), serialized));
+ }
+ }
+ }
+
+ // Second, commit data
+ try (LakeCommitter<LanceWriteResult, LanceCommittable> lakeCommitter =
+ createLakeCommitter(tablePath, tableInfo)) {
+ // serialize/deserialize committable
+ LanceCommittable lanceCommittable = lakeCommitter.toCommittable(lanceWriteResults);
+ byte[] serialized = committableSerializer.serialize(lanceCommittable);
+ lanceCommittable =
+ committableSerializer.deserialize(
+ committableSerializer.getVersion(), serialized);
+ Map<String, String> snapshotProperties =
+ Collections.singletonMap(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets");
+ LakeCommitResult commitResult =
+ lakeCommitter.commit(lanceCommittable, snapshotProperties);
+ // lance dataset version starts from 1
+ assertThat(commitResult.getCommittedSnapshotId()).isEqualTo(2);
+ }
+
+ try (Dataset dataset =
+ Dataset.open(
+ new RootAllocator(),
+ config.getDatasetUri(),
+ LanceConfig.genReadOptionFromConfig(config))) {
+ ArrowReader reader = dataset.newScan().scanBatches();
+ VectorSchemaRoot readerRoot = reader.getVectorSchemaRoot();
+
+ // Verify data can be read back
+ for (int bucket = 0; bucket < 3; bucket++) {
+ for (String partition : partitionIdAndName.values()) {
+ reader.loadNextBatch();
+ Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
+ List<LogRecord> expectRecords = recordsByBucket.get(partitionBucket);
+ verifyNestedRowRecords(readerRoot, expectRecords, bucket, isPartitioned);
+ }
+ }
+ assertThat(reader.loadNextBatch()).isFalse();
+ }
+ }
+
+ private Schema createNestedRowTable(LanceConfig config) {
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column(
+ "address",
+ DataTypes.ROW(
+ DataTypes.FIELD("city", DataTypes.STRING()),
+ DataTypes.FIELD("zip", DataTypes.INT())));
+ Schema schema = schemaBuilder.build();
+ WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
+ LanceDatasetAdapter.createDataset(
+ config.getDatasetUri(), LanceArrowUtils.toArrowSchema(schema.getRowType()), params);
+
+ return schema;
+ }
+
+ private Tuple2<List<LogRecord>, List<LogRecord>> genNestedRowLogRecords(
+ @Nullable String partition, int bucket, int numRecords) {
+ List<LogRecord> logRecords = new ArrayList<>();
+ for (int i = 0; i < numRecords; i++) {
+ GenericRow genericRow = new GenericRow(3);
+ genericRow.setField(0, i);
+ genericRow.setField(1, BinaryString.fromString("user" + bucket + "_" + i));
+
+ // Create nested address row
+ GenericRow addressRow = new GenericRow(2);
+ addressRow.setField(0, BinaryString.fromString("city" + bucket));
+ addressRow.setField(1, 10000 + bucket);
+ genericRow.setField(2, addressRow);
+
+ LogRecord logRecord =
+ new GenericRecord(
+ i, System.currentTimeMillis(), ChangeType.APPEND_ONLY, genericRow);
+ logRecords.add(logRecord);
+ }
+ return Tuple2.of(logRecords, logRecords);
+ }
+
+ private void verifyNestedRowRecords(
+ VectorSchemaRoot root,
+ List<LogRecord> expectRecords,
+ int expectBucket,
+ boolean isPartitioned)
Review Comment:
Remove unused args.
##########
fluss-lake/fluss-lake-lance/src/test/java/org/apache/fluss/lake/lance/tiering/LanceTieringTest.java:
##########
@@ -355,4 +356,170 @@ private Schema createTable(LanceConfig config, Map<String, String> customPropert
return schema;
}
+
+ @ParameterizedTest
+ @MethodSource("tieringWriteArgs")
+ void testTieringWriteTableWithNestedRowType(boolean isPartitioned) throws Exception {
+ int bucketNum = 3;
+ TablePath tablePath = TablePath.of("lance", "nestedRowTable");
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put("lance.batch_size", "256");
+ LanceConfig config =
+ LanceConfig.from(
+ configuration.toMap(),
+ customProperties,
+ tablePath.getDatabaseName(),
+ tablePath.getTableName());
+ Schema schema = createNestedRowTable(config);
+
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(schema)
+ .distributedBy(bucketNum)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .customProperties(customProperties)
+ .build();
+ TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
+ List<LanceWriteResult> lanceWriteResults = new ArrayList<>();
+ SimpleVersionedSerializer<LanceWriteResult> writeResultSerializer =
+ lanceLakeTieringFactory.getWriteResultSerializer();
+ SimpleVersionedSerializer<LanceCommittable> committableSerializer =
+ lanceLakeTieringFactory.getCommittableSerializer();
+
+ Map<Tuple2<String, Integer>, List<LogRecord>> recordsByBucket = new HashMap<>();
+ Map<Long, String> partitionIdAndName =
+ isPartitioned
+ ? new HashMap<Long, String>() {
+ {
+ put(1L, "p1");
+ put(2L, "p2");
+ put(3L, "p3");
+ }
+ }
+ : Collections.singletonMap(null, null);
+
+ // First, write data with nested row types
+ for (int bucket = 0; bucket < bucketNum; bucket++) {
+ for (Map.Entry<Long, String> entry : partitionIdAndName.entrySet()) {
+ String partition = entry.getValue();
+ try (LakeWriter<LanceWriteResult> lakeWriter =
+ createLakeWriter(tablePath, bucket, partition, tableInfo)) {
+ Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
+ Tuple2<List<LogRecord>, List<LogRecord>> writeAndExpectRecords =
+ genNestedRowLogRecords(partition, bucket, 10);
+ List<LogRecord> writtenRecords = writeAndExpectRecords.f0;
+ List<LogRecord> expectRecords = writeAndExpectRecords.f1;
+ recordsByBucket.put(partitionBucket, expectRecords);
+ for (LogRecord logRecord : writtenRecords) {
+ lakeWriter.write(logRecord);
+ }
+ // serialize/deserialize writeResult
+ LanceWriteResult lanceWriteResult = lakeWriter.complete();
+ byte[] serialized = writeResultSerializer.serialize(lanceWriteResult);
+ lanceWriteResults.add(
+ writeResultSerializer.deserialize(
+ writeResultSerializer.getVersion(), serialized));
+ }
+ }
+ }
+
+ // Second, commit data
+ try (LakeCommitter<LanceWriteResult, LanceCommittable> lakeCommitter =
+ createLakeCommitter(tablePath, tableInfo)) {
+ // serialize/deserialize committable
+ LanceCommittable lanceCommittable = lakeCommitter.toCommittable(lanceWriteResults);
+ byte[] serialized = committableSerializer.serialize(lanceCommittable);
+ lanceCommittable =
+ committableSerializer.deserialize(
+ committableSerializer.getVersion(), serialized);
+ Map<String, String> snapshotProperties =
+ Collections.singletonMap(FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY, "offsets");
+ LakeCommitResult commitResult =
+ lakeCommitter.commit(lanceCommittable, snapshotProperties);
+ // lance dataset version starts from 1
+ assertThat(commitResult.getCommittedSnapshotId()).isEqualTo(2);
+ }
+
+ try (Dataset dataset =
+ Dataset.open(
+ new RootAllocator(),
+ config.getDatasetUri(),
+ LanceConfig.genReadOptionFromConfig(config))) {
+ ArrowReader reader = dataset.newScan().scanBatches();
+ VectorSchemaRoot readerRoot = reader.getVectorSchemaRoot();
+
+ // Verify data can be read back
+ for (int bucket = 0; bucket < 3; bucket++) {
+ for (String partition : partitionIdAndName.values()) {
+ reader.loadNextBatch();
+ Tuple2<String, Integer> partitionBucket = Tuple2.of(partition, bucket);
+ List<LogRecord> expectRecords = recordsByBucket.get(partitionBucket);
+ verifyNestedRowRecords(readerRoot, expectRecords, bucket, isPartitioned);
+ }
+ }
+ assertThat(reader.loadNextBatch()).isFalse();
+ }
+ }
+
+ private Schema createNestedRowTable(LanceConfig config) {
+ Schema.Builder schemaBuilder =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("name", DataTypes.STRING())
+ .column(
+ "address",
+ DataTypes.ROW(
+ DataTypes.FIELD("city", DataTypes.STRING()),
+ DataTypes.FIELD("zip", DataTypes.INT())));
+ Schema schema = schemaBuilder.build();
+ WriteParams params = LanceConfig.genWriteParamsFromConfig(config);
+ LanceDatasetAdapter.createDataset(
+ config.getDatasetUri(), LanceArrowUtils.toArrowSchema(schema.getRowType()), params);
+
+ return schema;
+ }
+
+ private Tuple2<List<LogRecord>, List<LogRecord>> genNestedRowLogRecords(
+ @Nullable String partition, int bucket, int numRecords) {
+ List<LogRecord> logRecords = new ArrayList<>();
+ for (int i = 0; i < numRecords; i++) {
+ GenericRow genericRow = new GenericRow(3);
+ genericRow.setField(0, i);
+ genericRow.setField(1, BinaryString.fromString("user" + bucket + "_" + i));
+
+ // Create nested address row
+ GenericRow addressRow = new GenericRow(2);
+ addressRow.setField(0, BinaryString.fromString("city" + bucket));
+ addressRow.setField(1, 10000 + bucket);
+ genericRow.setField(2, addressRow);
+
+ LogRecord logRecord =
+ new GenericRecord(
+ i, System.currentTimeMillis(), ChangeType.APPEND_ONLY, genericRow);
+ logRecords.add(logRecord);
+ }
+ return Tuple2.of(logRecords, logRecords);
+ }
+
+ private void verifyNestedRowRecords(
+ VectorSchemaRoot root,
+ List<LogRecord> expectRecords,
+ int expectBucket,
+ boolean isPartitioned)
+ throws Exception {
Review Comment:
No method calls throw Exceptions, remove?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add NestedRow type support for Lance [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction merged PR #2578:
URL: https://github.com/apache/fluss/pull/2578
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add NestedRow type support for Lance [fluss]
Posted by "XuQianJin-Stars (via GitHub)" <gi...@apache.org>.
XuQianJin-Stars commented on PR #2578:
URL: https://github.com/apache/fluss/pull/2578#issuecomment-4115018830
> Hello, looks like mdx file change crept back in again. I have removed the unrelated change.
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add NestedRow type support for Lance [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #2578:
URL: https://github.com/apache/fluss/pull/2578#issuecomment-4100609147
LGTM overall, I'm away atm, will manual test/review on Sunday when I get back before approving/merging
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add NestedRow type support for Lance [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #2578:
URL: https://github.com/apache/fluss/pull/2578#issuecomment-4109058155
Hello, looks like mdx file change crept back in again. I have removed the unrelated change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/lance] Add NestedRow type support for Lance [fluss]
Posted by "XuQianJin-Stars (via GitHub)" <gi...@apache.org>.
XuQianJin-Stars commented on PR #2578:
URL: https://github.com/apache/fluss/pull/2578#issuecomment-4087446566
@leekeiabstraction Hi, i already updated the pr. Please help review when you got some time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org