You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "polyzos (via GitHub)" <gi...@apache.org> on 2025/11/17 14:41:37 UTC
[PR] [Client] Add Pojo Support [fluss]
polyzos opened a new pull request, #1992:
URL: https://github.com/apache/fluss/pull/1992
(no comment)
--
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] [Client] Add Pojo Support [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3575819560
@leekeiabstraction . Indeed, we are looking at a 2x performance penalty.
I tested writing/scanning on both PK and Log tables with 10 million records.
I ran 5 iterations for each and calculated the average.
This is basically a trade-off for the users. Using InternalRow/GenericRow directly is way more efficient; however, this might come with some extra complexity and boilerplatete code.
For this reason, I want to give flexibility, probably leave the docs as is, with GenericRow being the go-to approach, but also add a section that Pojos can be used directly and maybe highlight this trade-off.
Moving forward I'm thinking that maybe it makes sense to add some helper classes that also derive the schema for the table from a Pojo.
**Log Table**
<img width="726" height="1046" alt="Screenshot 2025-11-25 at 3 32 34 PM" src="https://github.com/user-attachments/assets/f955e85c-d5b6-4a1b-86ed-13f8113a68a8" />
**Primary Key Table**
<img width="749" height="1058" alt="Screenshot 2025-11-25 at 3 51 46 PM" src="https://github.com/user-attachments/assets/ba8a1ae3-29ac-4d1e-b2c0-922fd405c9a0" />
--
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] [Client] Add Pojo Support [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3678912285
@wuchong thank you for all your comments.
I have already introduced typed classes, like
--
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] [Client] Add Pojo Support [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #1992:
URL: https://github.com/apache/fluss/pull/1992#discussion_r2642891636
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TableUpsert.java:
##########
@@ -95,4 +95,11 @@ public Upsert partialUpdate(String... targetColumnNames) {
public UpsertWriter createWriter() {
return new UpsertWriterImpl(tablePath, tableInfo, targetColumns, writerClient);
}
+
+ @Override
+ public <T> TypedUpsertWriter<T> createTypedWriter(Class<T> pojoClass) {
+ UpsertWriterImpl delegate =
+ new UpsertWriterImpl(tablePath, tableInfo, targetColumns, writerClient);
Review Comment:
Can simplify to just call `createWriter()`.
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TableAppend.java:
##########
@@ -38,4 +38,10 @@ public TableAppend(TablePath tablePath, TableInfo tableInfo, WriterClient writer
public AppendWriter createWriter() {
return new AppendWriterImpl(tablePath, tableInfo, writerClient);
}
+
+ @Override
+ public <T> TypedAppendWriter<T> createTypedWriter(Class<T> pojoClass) {
+ AppendWriterImpl delegate = new AppendWriterImpl(tablePath, tableInfo, writerClient);
Review Comment:
Can simplify to just call `createWriter()`.
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedUpsertWriterImpl.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link UpsertWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedUpsertWriterImpl<T> implements TypedUpsertWriter<T> {
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ private final UpsertWriterImpl delegate;
+
+ @Override
+ public void close() throws Exception {
+ delegate.close();
+ }
+
+ private final Class<T> pojoClass;
Review Comment:
Not used, can be removed.
##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTypedClientITCase.java:
##########
@@ -0,0 +1,452 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table;
+
+import org.apache.fluss.client.admin.ClientToServerITCaseBase;
+import org.apache.fluss.client.converter.RowToPojoConverter;
+import org.apache.fluss.client.lookup.LookupResult;
+import org.apache.fluss.client.lookup.Lookuper;
+import org.apache.fluss.client.lookup.TypedLookuper;
+import org.apache.fluss.client.table.scanner.Scan;
+import org.apache.fluss.client.table.scanner.TypedScanRecord;
+import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
+import org.apache.fluss.client.table.scanner.log.TypedScanRecords;
+import org.apache.fluss.client.table.writer.TypedAppendWriter;
+import org.apache.fluss.client.table.writer.TypedUpsertWriter;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for writing and scanning POJOs via client API. */
+public class FlussTypedClientITCase extends ClientToServerITCaseBase {
+
+ /** Test POJO containing all supported field types used by converters. */
+ public static class AllTypesPojo {
+ // primary key
+ public Integer a;
+ // all supported converter fields
+ public Boolean bool1;
+ public Byte tiny;
+ public Short small;
+ public Integer intv;
+ public Long big;
+ public Float flt;
+ public Double dbl;
+ public Character ch;
+ public String str;
+ public byte[] bin;
+ public byte[] bytes;
+ public BigDecimal dec;
+ public LocalDate dt;
+ public LocalTime tm;
+ public LocalDateTime tsNtz;
+ public Instant tsLtz;
+
+ public AllTypesPojo() {}
+
+ public AllTypesPojo(
+ Integer a,
+ Boolean bool1,
+ Byte tiny,
+ Short small,
+ Integer intv,
+ Long big,
+ Float flt,
+ Double dbl,
+ Character ch,
+ String str,
+ byte[] bin,
+ byte[] bytes,
+ BigDecimal dec,
+ LocalDate dt,
+ LocalTime tm,
+ LocalDateTime tsNtz,
+ Instant tsLtz) {
+ this.a = a;
+ this.bool1 = bool1;
+ this.tiny = tiny;
+ this.small = small;
+ this.intv = intv;
+ this.big = big;
+ this.flt = flt;
+ this.dbl = dbl;
+ this.ch = ch;
+ this.str = str;
+ this.bin = bin;
+ this.bytes = bytes;
+ this.dec = dec;
+ this.dt = dt;
+ this.tm = tm;
+ this.tsNtz = tsNtz;
+ this.tsLtz = tsLtz;
+ }
+ }
+
+ /** Minimal POJO representing the primary key for {@link AllTypesPojo}. */
+ public static class PLookupKey {
+ public Integer a;
+
+ public PLookupKey() {}
+
+ public PLookupKey(Integer a) {
+ this.a = a;
+ }
+ }
+
+ private static Schema allTypesLogSchema() {
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ }
+
+ private static Schema allTypesPkSchema() {
+ // Same columns as log schema but with PK on 'a'
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .primaryKey("a")
+ .build();
+ }
+
+ private static AllTypesPojo newAllTypesPojo(int i) {
+ Integer a = i;
+ Boolean bool1 = (i % 2) == 0;
+ Byte tiny = (byte) (i - 5);
+ Short small = (short) (100 + i);
+ Integer intv = 1000 + i;
+ Long big = 100000L + i;
+ Float flt = 1.5f + i;
+ Double dbl = 2.5 + i;
+ Character ch = (char) ('a' + (i % 26));
+ String str = "s" + i;
+ byte[] bin = new byte[] {(byte) i, (byte) (i + 1), (byte) (i + 2)};
+ byte[] bytes = new byte[] {(byte) (10 + i), (byte) (20 + i)};
+ BigDecimal dec = new BigDecimal("12345." + (10 + i)).setScale(2, RoundingMode.HALF_UP);
+ LocalDate dt = LocalDate.of(2024, 1, 1).plusDays(i);
+ LocalTime tm = LocalTime.of(12, (i * 7) % 60, (i * 11) % 60);
+ LocalDateTime tsNtz = LocalDateTime.of(2024, 1, 1, 0, 0).plusSeconds(i).withNano(0);
+ Instant tsLtz = Instant.ofEpochMilli(1700000000000L + (i * 1000L));
+ return new AllTypesPojo(
+ a, bool1, tiny, small, intv, big, flt, dbl, ch, str, bin, bytes, dec, dt, tm, tsNtz,
+ tsLtz);
+ }
+
+ @Test
+ void testTypedAppendWriteAndScan() throws Exception {
+ // Build all-types log table schema
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ TablePath path = TablePath.of("pojo_db", "all_types_log");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2).build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ // write
+ TypedAppendWriter<AllTypesPojo> writer =
+ table.newAppend().createTypedWriter(AllTypesPojo.class);
+ List<AllTypesPojo> expected = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ AllTypesPojo u = newAllTypesPojo(i);
+ expected.add(u);
+ writer.append(u).get();
+ }
+ writer.flush();
+
+ // read
+ Scan scan = table.newScan();
+ TypedLogScanner<AllTypesPojo> scanner = scan.createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<AllTypesPojo> actual = new ArrayList<>();
+ while (actual.size() < expected.size()) {
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ assertThat(r.getChangeType()).isEqualTo(ChangeType.APPEND_ONLY);
+ actual.add(r.getValue());
+ }
+ }
+ assertThat(actual)
+ .usingRecursiveFieldByFieldElementComparator()
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+
+ @Test
+ void testTypedUpsertWriteAndScan() throws Exception {
+ // Build all-types PK table schema (PK on 'a')
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "all_types_pk");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ Upsert upsert = table.newUpsert();
+ TypedUpsertWriter<AllTypesPojo> writer = upsert.createTypedWriter(AllTypesPojo.class);
+
+ AllTypesPojo p1 = newAllTypesPojo(1);
+ AllTypesPojo p2 = newAllTypesPojo(2);
+ writer.upsert(p1).get();
+ writer.upsert(p2).get();
+
+ // update key 1: change a couple of fields
+ AllTypesPojo p1Updated = newAllTypesPojo(1);
+ p1Updated.str = "a1";
+ p1Updated.dec = new java.math.BigDecimal("42.42");
+ writer.upsert(p1Updated).get();
+ writer.flush();
+
+ // scan as POJOs and verify change types and values
+ TypedLogScanner<AllTypesPojo> scanner =
+ table.newScan().createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<ChangeType> changes = new ArrayList<>();
+ List<AllTypesPojo> values = new ArrayList<>();
+ while (values.size() < 4) { // INSERT 1, INSERT 2, UPDATE_BEFORE 1, UPDATE_AFTER 1
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ changes.add(r.getChangeType());
+ values.add(r.getValue());
+ }
+ }
+ assertThat(changes)
+ .contains(ChangeType.INSERT, ChangeType.UPDATE_BEFORE, ChangeType.UPDATE_AFTER);
+ // ensure the last update_after reflects new value
+ int lastIdx = changes.lastIndexOf(ChangeType.UPDATE_AFTER);
+ assertThat(values.get(lastIdx).str).isEqualTo("a1");
+ }
+ }
+
+ @Test
+ void testTypedLookups() throws Exception {
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "lookup_pk");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ TypedUpsertWriter<AllTypesPojo> writer =
+ table.newUpsert().createTypedWriter(AllTypesPojo.class);
+ writer.upsert(newAllTypesPojo(1)).get();
+ writer.upsert(newAllTypesPojo(2)).get();
+ writer.close();
+
+ // primary key lookup using Lookuper API with POJO key
+ TypedLookuper<PLookupKey> lookuper =
+ table.newLookup().createTypedLookuper(PLookupKey.class);
+ RowType tableSchema = table.getTableInfo().getRowType();
+ RowToPojoConverter<AllTypesPojo> rowConv =
+ RowToPojoConverter.of(AllTypesPojo.class, tableSchema, tableSchema);
+
+ LookupResult lr = lookuper.lookup(new PLookupKey(1)).get();
+ AllTypesPojo one = rowConv.fromRow(lr.getSingletonRow());
+ assertThat(one.str).isEqualTo("s1");
+ }
+ }
+
+ @Test
+ void testInternalRowLookup() throws Exception {
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "lookup_internalrow");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ // write a couple of rows via POJO writer
+ TypedUpsertWriter<AllTypesPojo> writer =
+ table.newUpsert().createTypedWriter(AllTypesPojo.class);
+ writer.upsert(newAllTypesPojo(101)).get();
+ writer.upsert(newAllTypesPojo(202)).get();
+ writer.close();
+
+ // now perform lookup using the raw InternalRow path to ensure it's still supported
+ Lookuper lookuper = table.newLookup().createLookuper();
+ RowType tableSchema = table.getTableInfo().getRowType();
+ RowType keyProjection = tableSchema.project(table.getTableInfo().getPrimaryKeys());
+
+ // Build the key row directly using GenericRow to avoid any POJO conversion
+ GenericRow keyRow = new GenericRow(keyProjection.getFieldCount());
+ keyRow.setField(0, 101); // primary key field 'a'
+
+ LookupResult lr = lookuper.lookup(keyRow).get();
+ RowToPojoConverter<AllTypesPojo> rowConv =
+ RowToPojoConverter.of(AllTypesPojo.class, tableSchema, tableSchema);
+ AllTypesPojo pojo = rowConv.fromRow(lr.getSingletonRow());
+ assertThat(pojo).isNotNull();
+ assertThat(pojo.a).isEqualTo(101);
+ assertThat(pojo.str).isEqualTo("s101");
+ }
+ }
+
+ @Test
+ void testTypedProjections() throws Exception {
+ TablePath path = TablePath.of("pojo_db", "proj_log");
+ TableDescriptor td =
+ TableDescriptor.builder().schema(allTypesLogSchema()).distributedBy(1).build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ TypedAppendWriter<AllTypesPojo> writer =
+ table.newAppend().createTypedWriter(AllTypesPojo.class);
+ writer.append(newAllTypesPojo(10)).get();
+ writer.append(newAllTypesPojo(11)).get();
+ writer.flush();
+
+ // Project only a subset of fields
+ TypedLogScanner<AllTypesPojo> scanner =
+ table.newScan()
+ .project(Arrays.asList("a", "str"))
+ .createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ AllTypesPojo u = r.getValue();
+ assertThat(u.a).isNotNull();
+ assertThat(u.str).isNotNull();
+ // non-projected fields should be null
+ assertThat(u.bool1).isNull();
+ assertThat(u.bin).isNull();
+ assertThat(u.bytes).isNull();
+ assertThat(u.dec).isNull();
+ assertThat(u.dt).isNull();
+ assertThat(u.tm).isNull();
+ assertThat(u.tsNtz).isNull();
+ assertThat(u.tsLtz).isNull();
+ }
+ }
+ }
+
+ @Test
+ void testTypedPartialUpdates() throws Exception {
+ // Use full PK schema and update a subset of fields
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "pk_partial");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(1, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ Upsert upsert = table.newUpsert().partialUpdate("a", "str", "dec");
+ TypedUpsertWriter<AllTypesPojo> writer = upsert.createTypedWriter(AllTypesPojo.class);
+
+ // initial full row
+ writer.upsert(newAllTypesPojo(1)).get();
+
+ // partial update: only PK + subset fields
+ AllTypesPojo patch = new AllTypesPojo();
+ patch.a = 1;
+ patch.str = "second";
+ patch.dec = new BigDecimal("99.99");
+ writer.upsert(patch).get();
+ writer.close();
+
+ // verify via lookup and scan using Lookuper + POJO key
+ TypedLookuper<PLookupKey> lookuper =
+ table.newLookup().createTypedLookuper(PLookupKey.class);
+ RowType tableSchema = table.getTableInfo().getRowType();
+ RowToPojoConverter<AllTypesPojo> rowConv =
+ RowToPojoConverter.of(AllTypesPojo.class, tableSchema, tableSchema);
+ AllTypesPojo lookedUp =
+ rowConv.fromRow(lookuper.lookup(new PLookupKey(1)).get().getSingletonRow());
+ assertThat(lookedUp.str).isEqualTo("second");
+ assertThat(lookedUp.dec).isEqualByComparingTo("99.99");
+
+ TypedLogScanner<AllTypesPojo> scanner =
+ table.newScan().createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+ boolean sawUpdateAfter = false;
+ while (!sawUpdateAfter) {
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ if (r.getChangeType() == ChangeType.UPDATE_AFTER) {
+ assertThat(r.getValue().str).isEqualTo("second");
Review Comment:
ditto
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedUpsertWriterImpl.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link UpsertWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedUpsertWriterImpl<T> implements TypedUpsertWriter<T> {
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ private final UpsertWriterImpl delegate;
+
+ @Override
+ public void close() throws Exception {
+ delegate.close();
+ }
+
+ private final Class<T> pojoClass;
+ private final TableInfo tableInfo;
+ private final RowType tableSchema;
+ private final int[] targetColumns; // may be null
Review Comment:
We can add `@Nullable` annotation to indicate it is nullable
##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTypedClientITCase.java:
##########
@@ -0,0 +1,452 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table;
+
+import org.apache.fluss.client.admin.ClientToServerITCaseBase;
+import org.apache.fluss.client.converter.RowToPojoConverter;
+import org.apache.fluss.client.lookup.LookupResult;
+import org.apache.fluss.client.lookup.Lookuper;
+import org.apache.fluss.client.lookup.TypedLookuper;
+import org.apache.fluss.client.table.scanner.Scan;
+import org.apache.fluss.client.table.scanner.TypedScanRecord;
+import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
+import org.apache.fluss.client.table.scanner.log.TypedScanRecords;
+import org.apache.fluss.client.table.writer.TypedAppendWriter;
+import org.apache.fluss.client.table.writer.TypedUpsertWriter;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for writing and scanning POJOs via client API. */
+public class FlussTypedClientITCase extends ClientToServerITCaseBase {
+
+ /** Test POJO containing all supported field types used by converters. */
+ public static class AllTypesPojo {
+ // primary key
+ public Integer a;
+ // all supported converter fields
+ public Boolean bool1;
+ public Byte tiny;
+ public Short small;
+ public Integer intv;
+ public Long big;
+ public Float flt;
+ public Double dbl;
+ public Character ch;
+ public String str;
+ public byte[] bin;
+ public byte[] bytes;
+ public BigDecimal dec;
+ public LocalDate dt;
+ public LocalTime tm;
+ public LocalDateTime tsNtz;
+ public Instant tsLtz;
+
+ public AllTypesPojo() {}
+
+ public AllTypesPojo(
+ Integer a,
+ Boolean bool1,
+ Byte tiny,
+ Short small,
+ Integer intv,
+ Long big,
+ Float flt,
+ Double dbl,
+ Character ch,
+ String str,
+ byte[] bin,
+ byte[] bytes,
+ BigDecimal dec,
+ LocalDate dt,
+ LocalTime tm,
+ LocalDateTime tsNtz,
+ Instant tsLtz) {
+ this.a = a;
+ this.bool1 = bool1;
+ this.tiny = tiny;
+ this.small = small;
+ this.intv = intv;
+ this.big = big;
+ this.flt = flt;
+ this.dbl = dbl;
+ this.ch = ch;
+ this.str = str;
+ this.bin = bin;
+ this.bytes = bytes;
+ this.dec = dec;
+ this.dt = dt;
+ this.tm = tm;
+ this.tsNtz = tsNtz;
+ this.tsLtz = tsLtz;
+ }
+ }
+
+ /** Minimal POJO representing the primary key for {@link AllTypesPojo}. */
+ public static class PLookupKey {
+ public Integer a;
+
+ public PLookupKey() {}
+
+ public PLookupKey(Integer a) {
+ this.a = a;
+ }
+ }
+
+ private static Schema allTypesLogSchema() {
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ }
+
+ private static Schema allTypesPkSchema() {
+ // Same columns as log schema but with PK on 'a'
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .primaryKey("a")
+ .build();
+ }
+
+ private static AllTypesPojo newAllTypesPojo(int i) {
+ Integer a = i;
+ Boolean bool1 = (i % 2) == 0;
+ Byte tiny = (byte) (i - 5);
+ Short small = (short) (100 + i);
+ Integer intv = 1000 + i;
+ Long big = 100000L + i;
+ Float flt = 1.5f + i;
+ Double dbl = 2.5 + i;
+ Character ch = (char) ('a' + (i % 26));
+ String str = "s" + i;
+ byte[] bin = new byte[] {(byte) i, (byte) (i + 1), (byte) (i + 2)};
+ byte[] bytes = new byte[] {(byte) (10 + i), (byte) (20 + i)};
+ BigDecimal dec = new BigDecimal("12345." + (10 + i)).setScale(2, RoundingMode.HALF_UP);
+ LocalDate dt = LocalDate.of(2024, 1, 1).plusDays(i);
+ LocalTime tm = LocalTime.of(12, (i * 7) % 60, (i * 11) % 60);
+ LocalDateTime tsNtz = LocalDateTime.of(2024, 1, 1, 0, 0).plusSeconds(i).withNano(0);
+ Instant tsLtz = Instant.ofEpochMilli(1700000000000L + (i * 1000L));
+ return new AllTypesPojo(
+ a, bool1, tiny, small, intv, big, flt, dbl, ch, str, bin, bytes, dec, dt, tm, tsNtz,
+ tsLtz);
+ }
+
+ @Test
+ void testTypedAppendWriteAndScan() throws Exception {
+ // Build all-types log table schema
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ TablePath path = TablePath.of("pojo_db", "all_types_log");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2).build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ // write
+ TypedAppendWriter<AllTypesPojo> writer =
+ table.newAppend().createTypedWriter(AllTypesPojo.class);
+ List<AllTypesPojo> expected = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ AllTypesPojo u = newAllTypesPojo(i);
+ expected.add(u);
+ writer.append(u).get();
+ }
+ writer.flush();
+
+ // read
+ Scan scan = table.newScan();
+ TypedLogScanner<AllTypesPojo> scanner = scan.createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<AllTypesPojo> actual = new ArrayList<>();
+ while (actual.size() < expected.size()) {
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ assertThat(r.getChangeType()).isEqualTo(ChangeType.APPEND_ONLY);
+ actual.add(r.getValue());
+ }
+ }
+ assertThat(actual)
+ .usingRecursiveFieldByFieldElementComparator()
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+
+ @Test
+ void testTypedUpsertWriteAndScan() throws Exception {
+ // Build all-types PK table schema (PK on 'a')
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "all_types_pk");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ Upsert upsert = table.newUpsert();
+ TypedUpsertWriter<AllTypesPojo> writer = upsert.createTypedWriter(AllTypesPojo.class);
+
+ AllTypesPojo p1 = newAllTypesPojo(1);
+ AllTypesPojo p2 = newAllTypesPojo(2);
+ writer.upsert(p1).get();
+ writer.upsert(p2).get();
+
+ // update key 1: change a couple of fields
+ AllTypesPojo p1Updated = newAllTypesPojo(1);
+ p1Updated.str = "a1";
+ p1Updated.dec = new java.math.BigDecimal("42.42");
+ writer.upsert(p1Updated).get();
+ writer.flush();
+
+ // scan as POJOs and verify change types and values
+ TypedLogScanner<AllTypesPojo> scanner =
+ table.newScan().createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<ChangeType> changes = new ArrayList<>();
+ List<AllTypesPojo> values = new ArrayList<>();
+ while (values.size() < 4) { // INSERT 1, INSERT 2, UPDATE_BEFORE 1, UPDATE_AFTER 1
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ changes.add(r.getChangeType());
+ values.add(r.getValue());
+ }
+ }
+ assertThat(changes)
+ .contains(ChangeType.INSERT, ChangeType.UPDATE_BEFORE, ChangeType.UPDATE_AFTER);
Review Comment:
```suggestion
assertThat(changes)
.containsExactlyInAnyOrder(
ChangeType.INSERT,
ChangeType.INSERT,
ChangeType.UPDATE_BEFORE,
ChangeType.UPDATE_AFTER);
```
use `containsExactlyInAnyOrder` to verify all the changes
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedAppendWriterImpl.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link AppendWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedAppendWriterImpl<T> implements TypedAppendWriter<T> {
+
+ private final AppendWriterImpl delegate;
+ private final Class<T> pojoClass;
+ private final RowType tableSchema;
Review Comment:
not used, can be removed.
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedUpsertWriterImpl.java:
##########
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link UpsertWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedUpsertWriterImpl<T> implements TypedUpsertWriter<T> {
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ private final UpsertWriterImpl delegate;
+
+ @Override
+ public void close() throws Exception {
+ delegate.close();
+ }
+
+ private final Class<T> pojoClass;
+ private final TableInfo tableInfo;
+ private final RowType tableSchema;
+ private final int[] targetColumns; // may be null
+
+ private final RowType pkProjection;
+ @Nullable private final RowType targetProjection;
+
+ private final PojoToRowConverter<T> pojoToRowConverter;
+ private final PojoToRowConverter<T> pkConverter;
+ @Nullable private final PojoToRowConverter<T> targetConverter;
+
+ TypedUpsertWriterImpl(
+ UpsertWriterImpl delegate,
+ Class<T> pojoClass,
+ TableInfo tableInfo,
+ int[] targetColumns) {
+ this.delegate = delegate;
+ this.pojoClass = pojoClass;
+ this.tableInfo = tableInfo;
+ this.tableSchema = tableInfo.getRowType();
+ this.targetColumns = targetColumns;
+
+ // Precompute projections
+ this.pkProjection = this.tableSchema.project(tableInfo.getPhysicalPrimaryKeys());
+ this.targetProjection =
+ (targetColumns == null) ? null : this.tableSchema.project(targetColumns);
+
+ // Initialize reusable converters
+ this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
+ this.pkConverter = PojoToRowConverter.of(pojoClass, tableSchema, pkProjection);
+ this.targetConverter =
+ (targetProjection == null)
+ ? null
+ : PojoToRowConverter.of(pojoClass, tableSchema, targetProjection);
+ }
+
+ @Override
+ public CompletableFuture<UpsertResult> upsert(T record) {
+ if (record instanceof InternalRow) {
+ return delegate.upsert((InternalRow) record);
+ }
+ InternalRow row = convertPojo(record, /*forDelete=*/ false);
+ return delegate.upsert(row);
+ }
+
+ @Override
+ public CompletableFuture<DeleteResult> delete(T record) {
+ if (record instanceof InternalRow) {
+ return delegate.delete((InternalRow) record);
+ }
+ InternalRow pkOnly = convertPojo(record, /*forDelete=*/ true);
+ return delegate.delete(pkOnly);
+ }
+
+ private InternalRow convertPojo(T pojo, boolean forDelete) {
+ final RowType projection;
+ final PojoToRowConverter<T> converter;
+ if (forDelete) {
+ projection = pkProjection;
+ converter = pkConverter;
+ } else if (targetProjection != null && targetConverter != null) {
+ projection = targetProjection;
+ converter = targetConverter;
+ } else {
+ projection = tableSchema;
+ converter = pojoToRowConverter;
+ }
+
+ GenericRow projected = converter.toRow(pojo);
+ if (projection == tableSchema) {
+ return projected;
+ }
+ // expand projected row to full row if needed
+ GenericRow full = new GenericRow(tableSchema.getFieldCount());
+ if (forDelete) {
+ // set PK fields, others null
+ for (String pk : tableInfo.getPhysicalPrimaryKeys()) {
+ int projIndex = projection.getFieldIndex(pk);
+ int fullIndex = tableSchema.getFieldIndex(pk);
Review Comment:
Could you please add a `TODO` comment above this code block noting that it can be optimized by pre-computing the index mapping in the constructor?
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java:
##########
@@ -37,7 +37,9 @@
*/
@PublicEvolving
public class ScanRecords implements Iterable<ScanRecord> {
- public static final ScanRecords EMPTY = new ScanRecords(Collections.emptyMap());
+ public static final ScanRecords empty() {
+ return new ScanRecords(Collections.emptyMap());
+ }
Review Comment:
Not necessary? I still prefer the preivous implementation because it can avoid some small object overhead (GC).
##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTypedClientITCase.java:
##########
@@ -0,0 +1,452 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table;
+
+import org.apache.fluss.client.admin.ClientToServerITCaseBase;
+import org.apache.fluss.client.converter.RowToPojoConverter;
+import org.apache.fluss.client.lookup.LookupResult;
+import org.apache.fluss.client.lookup.Lookuper;
+import org.apache.fluss.client.lookup.TypedLookuper;
+import org.apache.fluss.client.table.scanner.Scan;
+import org.apache.fluss.client.table.scanner.TypedScanRecord;
+import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
+import org.apache.fluss.client.table.scanner.log.TypedScanRecords;
+import org.apache.fluss.client.table.writer.TypedAppendWriter;
+import org.apache.fluss.client.table.writer.TypedUpsertWriter;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for writing and scanning POJOs via client API. */
+public class FlussTypedClientITCase extends ClientToServerITCaseBase {
+
+ /** Test POJO containing all supported field types used by converters. */
+ public static class AllTypesPojo {
+ // primary key
+ public Integer a;
+ // all supported converter fields
+ public Boolean bool1;
+ public Byte tiny;
+ public Short small;
+ public Integer intv;
+ public Long big;
+ public Float flt;
+ public Double dbl;
+ public Character ch;
+ public String str;
+ public byte[] bin;
+ public byte[] bytes;
+ public BigDecimal dec;
+ public LocalDate dt;
+ public LocalTime tm;
+ public LocalDateTime tsNtz;
+ public Instant tsLtz;
+
+ public AllTypesPojo() {}
+
+ public AllTypesPojo(
+ Integer a,
+ Boolean bool1,
+ Byte tiny,
+ Short small,
+ Integer intv,
+ Long big,
+ Float flt,
+ Double dbl,
+ Character ch,
+ String str,
+ byte[] bin,
+ byte[] bytes,
+ BigDecimal dec,
+ LocalDate dt,
+ LocalTime tm,
+ LocalDateTime tsNtz,
+ Instant tsLtz) {
+ this.a = a;
+ this.bool1 = bool1;
+ this.tiny = tiny;
+ this.small = small;
+ this.intv = intv;
+ this.big = big;
+ this.flt = flt;
+ this.dbl = dbl;
+ this.ch = ch;
+ this.str = str;
+ this.bin = bin;
+ this.bytes = bytes;
+ this.dec = dec;
+ this.dt = dt;
+ this.tm = tm;
+ this.tsNtz = tsNtz;
+ this.tsLtz = tsLtz;
+ }
+ }
+
+ /** Minimal POJO representing the primary key for {@link AllTypesPojo}. */
+ public static class PLookupKey {
+ public Integer a;
+
+ public PLookupKey() {}
+
+ public PLookupKey(Integer a) {
+ this.a = a;
+ }
+ }
+
+ private static Schema allTypesLogSchema() {
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ }
+
+ private static Schema allTypesPkSchema() {
+ // Same columns as log schema but with PK on 'a'
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .primaryKey("a")
+ .build();
+ }
+
+ private static AllTypesPojo newAllTypesPojo(int i) {
+ Integer a = i;
+ Boolean bool1 = (i % 2) == 0;
+ Byte tiny = (byte) (i - 5);
+ Short small = (short) (100 + i);
+ Integer intv = 1000 + i;
+ Long big = 100000L + i;
+ Float flt = 1.5f + i;
+ Double dbl = 2.5 + i;
+ Character ch = (char) ('a' + (i % 26));
+ String str = "s" + i;
+ byte[] bin = new byte[] {(byte) i, (byte) (i + 1), (byte) (i + 2)};
+ byte[] bytes = new byte[] {(byte) (10 + i), (byte) (20 + i)};
+ BigDecimal dec = new BigDecimal("12345." + (10 + i)).setScale(2, RoundingMode.HALF_UP);
+ LocalDate dt = LocalDate.of(2024, 1, 1).plusDays(i);
+ LocalTime tm = LocalTime.of(12, (i * 7) % 60, (i * 11) % 60);
+ LocalDateTime tsNtz = LocalDateTime.of(2024, 1, 1, 0, 0).plusSeconds(i).withNano(0);
+ Instant tsLtz = Instant.ofEpochMilli(1700000000000L + (i * 1000L));
+ return new AllTypesPojo(
+ a, bool1, tiny, small, intv, big, flt, dbl, ch, str, bin, bytes, dec, dt, tm, tsNtz,
+ tsLtz);
+ }
+
+ @Test
+ void testTypedAppendWriteAndScan() throws Exception {
+ // Build all-types log table schema
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ TablePath path = TablePath.of("pojo_db", "all_types_log");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2).build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ // write
+ TypedAppendWriter<AllTypesPojo> writer =
+ table.newAppend().createTypedWriter(AllTypesPojo.class);
+ List<AllTypesPojo> expected = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ AllTypesPojo u = newAllTypesPojo(i);
+ expected.add(u);
+ writer.append(u).get();
+ }
+ writer.flush();
+
+ // read
+ Scan scan = table.newScan();
+ TypedLogScanner<AllTypesPojo> scanner = scan.createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<AllTypesPojo> actual = new ArrayList<>();
+ while (actual.size() < expected.size()) {
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ assertThat(r.getChangeType()).isEqualTo(ChangeType.APPEND_ONLY);
+ actual.add(r.getValue());
+ }
+ }
+ assertThat(actual)
+ .usingRecursiveFieldByFieldElementComparator()
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+
+ @Test
+ void testTypedUpsertWriteAndScan() throws Exception {
+ // Build all-types PK table schema (PK on 'a')
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "all_types_pk");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ Upsert upsert = table.newUpsert();
+ TypedUpsertWriter<AllTypesPojo> writer = upsert.createTypedWriter(AllTypesPojo.class);
+
+ AllTypesPojo p1 = newAllTypesPojo(1);
+ AllTypesPojo p2 = newAllTypesPojo(2);
+ writer.upsert(p1).get();
+ writer.upsert(p2).get();
+
+ // update key 1: change a couple of fields
+ AllTypesPojo p1Updated = newAllTypesPojo(1);
+ p1Updated.str = "a1";
+ p1Updated.dec = new java.math.BigDecimal("42.42");
+ writer.upsert(p1Updated).get();
+ writer.flush();
+
+ // scan as POJOs and verify change types and values
+ TypedLogScanner<AllTypesPojo> scanner =
+ table.newScan().createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<ChangeType> changes = new ArrayList<>();
+ List<AllTypesPojo> values = new ArrayList<>();
+ while (values.size() < 4) { // INSERT 1, INSERT 2, UPDATE_BEFORE 1, UPDATE_AFTER 1
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ changes.add(r.getChangeType());
+ values.add(r.getValue());
+ }
+ }
+ assertThat(changes)
+ .contains(ChangeType.INSERT, ChangeType.UPDATE_BEFORE, ChangeType.UPDATE_AFTER);
+ // ensure the last update_after reflects new value
+ int lastIdx = changes.lastIndexOf(ChangeType.UPDATE_AFTER);
+ assertThat(values.get(lastIdx).str).isEqualTo("a1");
+ }
+ }
+
+ @Test
+ void testTypedLookups() throws Exception {
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "lookup_pk");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ TypedUpsertWriter<AllTypesPojo> writer =
+ table.newUpsert().createTypedWriter(AllTypesPojo.class);
+ writer.upsert(newAllTypesPojo(1)).get();
+ writer.upsert(newAllTypesPojo(2)).get();
+ writer.close();
+
+ // primary key lookup using Lookuper API with POJO key
+ TypedLookuper<PLookupKey> lookuper =
+ table.newLookup().createTypedLookuper(PLookupKey.class);
+ RowType tableSchema = table.getTableInfo().getRowType();
+ RowToPojoConverter<AllTypesPojo> rowConv =
+ RowToPojoConverter.of(AllTypesPojo.class, tableSchema, tableSchema);
+
+ LookupResult lr = lookuper.lookup(new PLookupKey(1)).get();
+ AllTypesPojo one = rowConv.fromRow(lr.getSingletonRow());
+ assertThat(one.str).isEqualTo("s1");
Review Comment:
```
assertThat(one).isEqualTo(newAllTypesPojo(1));
```
After adding `equals` and `hashcode` method to the `AllTypesPojo` class, we can simply assert the full record, this can check the full POJO record deserialization.
##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTypedClientITCase.java:
##########
@@ -0,0 +1,452 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table;
+
+import org.apache.fluss.client.admin.ClientToServerITCaseBase;
+import org.apache.fluss.client.converter.RowToPojoConverter;
+import org.apache.fluss.client.lookup.LookupResult;
+import org.apache.fluss.client.lookup.Lookuper;
+import org.apache.fluss.client.lookup.TypedLookuper;
+import org.apache.fluss.client.table.scanner.Scan;
+import org.apache.fluss.client.table.scanner.TypedScanRecord;
+import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
+import org.apache.fluss.client.table.scanner.log.TypedScanRecords;
+import org.apache.fluss.client.table.writer.TypedAppendWriter;
+import org.apache.fluss.client.table.writer.TypedUpsertWriter;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for writing and scanning POJOs via client API. */
+public class FlussTypedClientITCase extends ClientToServerITCaseBase {
+
+ /** Test POJO containing all supported field types used by converters. */
+ public static class AllTypesPojo {
+ // primary key
+ public Integer a;
+ // all supported converter fields
+ public Boolean bool1;
+ public Byte tiny;
+ public Short small;
+ public Integer intv;
+ public Long big;
+ public Float flt;
+ public Double dbl;
+ public Character ch;
+ public String str;
+ public byte[] bin;
+ public byte[] bytes;
+ public BigDecimal dec;
+ public LocalDate dt;
+ public LocalTime tm;
+ public LocalDateTime tsNtz;
+ public Instant tsLtz;
+
+ public AllTypesPojo() {}
+
+ public AllTypesPojo(
+ Integer a,
+ Boolean bool1,
+ Byte tiny,
+ Short small,
+ Integer intv,
+ Long big,
+ Float flt,
+ Double dbl,
+ Character ch,
+ String str,
+ byte[] bin,
+ byte[] bytes,
+ BigDecimal dec,
+ LocalDate dt,
+ LocalTime tm,
+ LocalDateTime tsNtz,
+ Instant tsLtz) {
+ this.a = a;
+ this.bool1 = bool1;
+ this.tiny = tiny;
+ this.small = small;
+ this.intv = intv;
+ this.big = big;
+ this.flt = flt;
+ this.dbl = dbl;
+ this.ch = ch;
+ this.str = str;
+ this.bin = bin;
+ this.bytes = bytes;
+ this.dec = dec;
+ this.dt = dt;
+ this.tm = tm;
+ this.tsNtz = tsNtz;
+ this.tsLtz = tsLtz;
+ }
+ }
+
+ /** Minimal POJO representing the primary key for {@link AllTypesPojo}. */
+ public static class PLookupKey {
+ public Integer a;
+
+ public PLookupKey() {}
+
+ public PLookupKey(Integer a) {
+ this.a = a;
+ }
+ }
+
+ private static Schema allTypesLogSchema() {
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ }
+
+ private static Schema allTypesPkSchema() {
+ // Same columns as log schema but with PK on 'a'
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .primaryKey("a")
+ .build();
+ }
+
+ private static AllTypesPojo newAllTypesPojo(int i) {
+ Integer a = i;
+ Boolean bool1 = (i % 2) == 0;
+ Byte tiny = (byte) (i - 5);
+ Short small = (short) (100 + i);
+ Integer intv = 1000 + i;
+ Long big = 100000L + i;
+ Float flt = 1.5f + i;
+ Double dbl = 2.5 + i;
+ Character ch = (char) ('a' + (i % 26));
+ String str = "s" + i;
+ byte[] bin = new byte[] {(byte) i, (byte) (i + 1), (byte) (i + 2)};
+ byte[] bytes = new byte[] {(byte) (10 + i), (byte) (20 + i)};
+ BigDecimal dec = new BigDecimal("12345." + (10 + i)).setScale(2, RoundingMode.HALF_UP);
+ LocalDate dt = LocalDate.of(2024, 1, 1).plusDays(i);
+ LocalTime tm = LocalTime.of(12, (i * 7) % 60, (i * 11) % 60);
+ LocalDateTime tsNtz = LocalDateTime.of(2024, 1, 1, 0, 0).plusSeconds(i).withNano(0);
+ Instant tsLtz = Instant.ofEpochMilli(1700000000000L + (i * 1000L));
+ return new AllTypesPojo(
+ a, bool1, tiny, small, intv, big, flt, dbl, ch, str, bin, bytes, dec, dt, tm, tsNtz,
+ tsLtz);
+ }
+
+ @Test
+ void testTypedAppendWriteAndScan() throws Exception {
+ // Build all-types log table schema
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ TablePath path = TablePath.of("pojo_db", "all_types_log");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2).build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ // write
+ TypedAppendWriter<AllTypesPojo> writer =
+ table.newAppend().createTypedWriter(AllTypesPojo.class);
+ List<AllTypesPojo> expected = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ AllTypesPojo u = newAllTypesPojo(i);
+ expected.add(u);
+ writer.append(u).get();
Review Comment:
```suggestion
writer.append(u);
```
use async flush to reduce test time
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TableWriter.java:
##########
@@ -38,4 +38,9 @@ public interface TableWriter {
* results in an error.
*/
void flush();
+
+ @Override
+ default void close() throws Exception {
+ // by default do nothing
+ }
Review Comment:
It seems the newly introduced `close()` method doesn’t currently have any meaningful work to perform, as the writers don’t hold any resources at this stage. I suggest holding off on introducing it for now.
Typically, a `close()` method should either **flush pending records** or **ensure all previously submitted requests have completed**—otherwise, it may mislead users into expecting cleanup or finalization behavior that isn’t actually implemented. This introduces some complex to this PR.
##########
fluss-client/src/test/java/org/apache/fluss/client/table/FlussTypedClientITCase.java:
##########
@@ -0,0 +1,452 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table;
+
+import org.apache.fluss.client.admin.ClientToServerITCaseBase;
+import org.apache.fluss.client.converter.RowToPojoConverter;
+import org.apache.fluss.client.lookup.LookupResult;
+import org.apache.fluss.client.lookup.Lookuper;
+import org.apache.fluss.client.lookup.TypedLookuper;
+import org.apache.fluss.client.table.scanner.Scan;
+import org.apache.fluss.client.table.scanner.TypedScanRecord;
+import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
+import org.apache.fluss.client.table.scanner.log.TypedScanRecords;
+import org.apache.fluss.client.table.writer.TypedAppendWriter;
+import org.apache.fluss.client.table.writer.TypedUpsertWriter;
+import org.apache.fluss.client.table.writer.Upsert;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** End-to-end tests for writing and scanning POJOs via client API. */
+public class FlussTypedClientITCase extends ClientToServerITCaseBase {
+
+ /** Test POJO containing all supported field types used by converters. */
+ public static class AllTypesPojo {
+ // primary key
+ public Integer a;
+ // all supported converter fields
+ public Boolean bool1;
+ public Byte tiny;
+ public Short small;
+ public Integer intv;
+ public Long big;
+ public Float flt;
+ public Double dbl;
+ public Character ch;
+ public String str;
+ public byte[] bin;
+ public byte[] bytes;
+ public BigDecimal dec;
+ public LocalDate dt;
+ public LocalTime tm;
+ public LocalDateTime tsNtz;
+ public Instant tsLtz;
+
+ public AllTypesPojo() {}
+
+ public AllTypesPojo(
+ Integer a,
+ Boolean bool1,
+ Byte tiny,
+ Short small,
+ Integer intv,
+ Long big,
+ Float flt,
+ Double dbl,
+ Character ch,
+ String str,
+ byte[] bin,
+ byte[] bytes,
+ BigDecimal dec,
+ LocalDate dt,
+ LocalTime tm,
+ LocalDateTime tsNtz,
+ Instant tsLtz) {
+ this.a = a;
+ this.bool1 = bool1;
+ this.tiny = tiny;
+ this.small = small;
+ this.intv = intv;
+ this.big = big;
+ this.flt = flt;
+ this.dbl = dbl;
+ this.ch = ch;
+ this.str = str;
+ this.bin = bin;
+ this.bytes = bytes;
+ this.dec = dec;
+ this.dt = dt;
+ this.tm = tm;
+ this.tsNtz = tsNtz;
+ this.tsLtz = tsLtz;
+ }
+ }
+
+ /** Minimal POJO representing the primary key for {@link AllTypesPojo}. */
+ public static class PLookupKey {
+ public Integer a;
+
+ public PLookupKey() {}
+
+ public PLookupKey(Integer a) {
+ this.a = a;
+ }
+ }
+
+ private static Schema allTypesLogSchema() {
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ }
+
+ private static Schema allTypesPkSchema() {
+ // Same columns as log schema but with PK on 'a'
+ return Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .primaryKey("a")
+ .build();
+ }
+
+ private static AllTypesPojo newAllTypesPojo(int i) {
+ Integer a = i;
+ Boolean bool1 = (i % 2) == 0;
+ Byte tiny = (byte) (i - 5);
+ Short small = (short) (100 + i);
+ Integer intv = 1000 + i;
+ Long big = 100000L + i;
+ Float flt = 1.5f + i;
+ Double dbl = 2.5 + i;
+ Character ch = (char) ('a' + (i % 26));
+ String str = "s" + i;
+ byte[] bin = new byte[] {(byte) i, (byte) (i + 1), (byte) (i + 2)};
+ byte[] bytes = new byte[] {(byte) (10 + i), (byte) (20 + i)};
+ BigDecimal dec = new BigDecimal("12345." + (10 + i)).setScale(2, RoundingMode.HALF_UP);
+ LocalDate dt = LocalDate.of(2024, 1, 1).plusDays(i);
+ LocalTime tm = LocalTime.of(12, (i * 7) % 60, (i * 11) % 60);
+ LocalDateTime tsNtz = LocalDateTime.of(2024, 1, 1, 0, 0).plusSeconds(i).withNano(0);
+ Instant tsLtz = Instant.ofEpochMilli(1700000000000L + (i * 1000L));
+ return new AllTypesPojo(
+ a, bool1, tiny, small, intv, big, flt, dbl, ch, str, bin, bytes, dec, dt, tm, tsNtz,
+ tsLtz);
+ }
+
+ @Test
+ void testTypedAppendWriteAndScan() throws Exception {
+ // Build all-types log table schema
+ Schema schema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT())
+ .column("bool1", DataTypes.BOOLEAN())
+ .column("tiny", DataTypes.TINYINT())
+ .column("small", DataTypes.SMALLINT())
+ .column("intv", DataTypes.INT())
+ .column("big", DataTypes.BIGINT())
+ .column("flt", DataTypes.FLOAT())
+ .column("dbl", DataTypes.DOUBLE())
+ .column("ch", DataTypes.CHAR(1))
+ .column("str", DataTypes.STRING())
+ .column("bin", DataTypes.BINARY(3))
+ .column("bytes", DataTypes.BYTES())
+ .column("dec", DataTypes.DECIMAL(10, 2))
+ .column("dt", DataTypes.DATE())
+ .column("tm", DataTypes.TIME())
+ .column("tsNtz", DataTypes.TIMESTAMP(3))
+ .column("tsLtz", DataTypes.TIMESTAMP_LTZ(3))
+ .build();
+ TablePath path = TablePath.of("pojo_db", "all_types_log");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2).build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ // write
+ TypedAppendWriter<AllTypesPojo> writer =
+ table.newAppend().createTypedWriter(AllTypesPojo.class);
+ List<AllTypesPojo> expected = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ AllTypesPojo u = newAllTypesPojo(i);
+ expected.add(u);
+ writer.append(u).get();
+ }
+ writer.flush();
+
+ // read
+ Scan scan = table.newScan();
+ TypedLogScanner<AllTypesPojo> scanner = scan.createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<AllTypesPojo> actual = new ArrayList<>();
+ while (actual.size() < expected.size()) {
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ assertThat(r.getChangeType()).isEqualTo(ChangeType.APPEND_ONLY);
+ actual.add(r.getValue());
+ }
+ }
+ assertThat(actual)
+ .usingRecursiveFieldByFieldElementComparator()
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+
+ @Test
+ void testTypedUpsertWriteAndScan() throws Exception {
+ // Build all-types PK table schema (PK on 'a')
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "all_types_pk");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ Upsert upsert = table.newUpsert();
+ TypedUpsertWriter<AllTypesPojo> writer = upsert.createTypedWriter(AllTypesPojo.class);
+
+ AllTypesPojo p1 = newAllTypesPojo(1);
+ AllTypesPojo p2 = newAllTypesPojo(2);
+ writer.upsert(p1).get();
+ writer.upsert(p2).get();
+
+ // update key 1: change a couple of fields
+ AllTypesPojo p1Updated = newAllTypesPojo(1);
+ p1Updated.str = "a1";
+ p1Updated.dec = new java.math.BigDecimal("42.42");
+ writer.upsert(p1Updated).get();
+ writer.flush();
+
+ // scan as POJOs and verify change types and values
+ TypedLogScanner<AllTypesPojo> scanner =
+ table.newScan().createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+
+ List<ChangeType> changes = new ArrayList<>();
+ List<AllTypesPojo> values = new ArrayList<>();
+ while (values.size() < 4) { // INSERT 1, INSERT 2, UPDATE_BEFORE 1, UPDATE_AFTER 1
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ changes.add(r.getChangeType());
+ values.add(r.getValue());
+ }
+ }
+ assertThat(changes)
+ .contains(ChangeType.INSERT, ChangeType.UPDATE_BEFORE, ChangeType.UPDATE_AFTER);
+ // ensure the last update_after reflects new value
+ int lastIdx = changes.lastIndexOf(ChangeType.UPDATE_AFTER);
+ assertThat(values.get(lastIdx).str).isEqualTo("a1");
+ }
+ }
+
+ @Test
+ void testTypedLookups() throws Exception {
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "lookup_pk");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ TypedUpsertWriter<AllTypesPojo> writer =
+ table.newUpsert().createTypedWriter(AllTypesPojo.class);
+ writer.upsert(newAllTypesPojo(1)).get();
+ writer.upsert(newAllTypesPojo(2)).get();
+ writer.close();
+
+ // primary key lookup using Lookuper API with POJO key
+ TypedLookuper<PLookupKey> lookuper =
+ table.newLookup().createTypedLookuper(PLookupKey.class);
+ RowType tableSchema = table.getTableInfo().getRowType();
+ RowToPojoConverter<AllTypesPojo> rowConv =
+ RowToPojoConverter.of(AllTypesPojo.class, tableSchema, tableSchema);
+
+ LookupResult lr = lookuper.lookup(new PLookupKey(1)).get();
+ AllTypesPojo one = rowConv.fromRow(lr.getSingletonRow());
+ assertThat(one.str).isEqualTo("s1");
+ }
+ }
+
+ @Test
+ void testInternalRowLookup() throws Exception {
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "lookup_internalrow");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(2, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ // write a couple of rows via POJO writer
+ TypedUpsertWriter<AllTypesPojo> writer =
+ table.newUpsert().createTypedWriter(AllTypesPojo.class);
+ writer.upsert(newAllTypesPojo(101)).get();
+ writer.upsert(newAllTypesPojo(202)).get();
+ writer.close();
+
+ // now perform lookup using the raw InternalRow path to ensure it's still supported
+ Lookuper lookuper = table.newLookup().createLookuper();
+ RowType tableSchema = table.getTableInfo().getRowType();
+ RowType keyProjection = tableSchema.project(table.getTableInfo().getPrimaryKeys());
+
+ // Build the key row directly using GenericRow to avoid any POJO conversion
+ GenericRow keyRow = new GenericRow(keyProjection.getFieldCount());
+ keyRow.setField(0, 101); // primary key field 'a'
+
+ LookupResult lr = lookuper.lookup(keyRow).get();
+ RowToPojoConverter<AllTypesPojo> rowConv =
+ RowToPojoConverter.of(AllTypesPojo.class, tableSchema, tableSchema);
+ AllTypesPojo pojo = rowConv.fromRow(lr.getSingletonRow());
+ assertThat(pojo).isNotNull();
+ assertThat(pojo.a).isEqualTo(101);
+ assertThat(pojo.str).isEqualTo("s101");
+ }
+ }
+
+ @Test
+ void testTypedProjections() throws Exception {
+ TablePath path = TablePath.of("pojo_db", "proj_log");
+ TableDescriptor td =
+ TableDescriptor.builder().schema(allTypesLogSchema()).distributedBy(1).build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ TypedAppendWriter<AllTypesPojo> writer =
+ table.newAppend().createTypedWriter(AllTypesPojo.class);
+ writer.append(newAllTypesPojo(10)).get();
+ writer.append(newAllTypesPojo(11)).get();
+ writer.flush();
+
+ // Project only a subset of fields
+ TypedLogScanner<AllTypesPojo> scanner =
+ table.newScan()
+ .project(Arrays.asList("a", "str"))
+ .createTypedLogScanner(AllTypesPojo.class);
+ subscribeFromBeginning(scanner, table);
+ TypedScanRecords<AllTypesPojo> recs = scanner.poll(Duration.ofSeconds(2));
+ for (TypedScanRecord<AllTypesPojo> r : recs) {
+ AllTypesPojo u = r.getValue();
+ assertThat(u.a).isNotNull();
+ assertThat(u.str).isNotNull();
+ // non-projected fields should be null
+ assertThat(u.bool1).isNull();
+ assertThat(u.bin).isNull();
+ assertThat(u.bytes).isNull();
+ assertThat(u.dec).isNull();
+ assertThat(u.dt).isNull();
+ assertThat(u.tm).isNull();
+ assertThat(u.tsNtz).isNull();
+ assertThat(u.tsLtz).isNull();
+ }
+ }
+ }
+
+ @Test
+ void testTypedPartialUpdates() throws Exception {
+ // Use full PK schema and update a subset of fields
+ Schema schema = allTypesPkSchema();
+ TablePath path = TablePath.of("pojo_db", "pk_partial");
+ TableDescriptor td = TableDescriptor.builder().schema(schema).distributedBy(1, "a").build();
+ createTable(path, td, true);
+
+ try (Table table = conn.getTable(path)) {
+ Upsert upsert = table.newUpsert().partialUpdate("a", "str", "dec");
+ TypedUpsertWriter<AllTypesPojo> writer = upsert.createTypedWriter(AllTypesPojo.class);
+
+ // initial full row
+ writer.upsert(newAllTypesPojo(1)).get();
+
+ // partial update: only PK + subset fields
+ AllTypesPojo patch = new AllTypesPojo();
+ patch.a = 1;
+ patch.str = "second";
+ patch.dec = new BigDecimal("99.99");
+ writer.upsert(patch).get();
+ writer.close();
+
+ // verify via lookup and scan using Lookuper + POJO key
+ TypedLookuper<PLookupKey> lookuper =
+ table.newLookup().createTypedLookuper(PLookupKey.class);
+ RowType tableSchema = table.getTableInfo().getRowType();
+ RowToPojoConverter<AllTypesPojo> rowConv =
+ RowToPojoConverter.of(AllTypesPojo.class, tableSchema, tableSchema);
+ AllTypesPojo lookedUp =
+ rowConv.fromRow(lookuper.lookup(new PLookupKey(1)).get().getSingletonRow());
+ assertThat(lookedUp.str).isEqualTo("second");
+ assertThat(lookedUp.dec).isEqualByComparingTo("99.99");
Review Comment:
In order to test the partial update feature, we should assert the other fields are keep unchanged here.
--
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] [Client] Add Pojo Support [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on code in PR #1992:
URL: https://github.com/apache/fluss/pull/1992#discussion_r2560830695
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedUpsertWriter.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link UpsertWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedUpsertWriter<T> implements UpsertWriter<T> {
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ private final UpsertWriterImpl delegate;
+ private final Class<T> pojoClass;
+ private final TableInfo tableInfo;
+ private final RowType tableSchema;
+ private final int[] targetColumns; // may be null
+
+ private final RowType pkProjection;
+ @Nullable private final RowType targetProjection;
+
+ private final PojoToRowConverter<T> pojoToRowConverter;
+ private final PojoToRowConverter<T> pkConverter;
+ @Nullable private final PojoToRowConverter<T> targetConverter;
+
+ TypedUpsertWriter(
+ UpsertWriterImpl delegate,
+ Class<T> pojoClass,
+ TableInfo tableInfo,
+ int[] targetColumns) {
+ this.delegate = delegate;
+ this.pojoClass = pojoClass;
+ this.tableInfo = tableInfo;
+ this.tableSchema = tableInfo.getRowType();
+ this.targetColumns = targetColumns;
+
+ // Precompute projections
+ this.pkProjection = this.tableSchema.project(tableInfo.getPhysicalPrimaryKeys());
+ this.targetProjection =
+ (targetColumns == null) ? null : this.tableSchema.project(targetColumns);
+
+ // Initialize reusable converters
+ this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
+ this.pkConverter = PojoToRowConverter.of(pojoClass, tableSchema, pkProjection);
+ this.targetConverter =
+ (targetProjection == null)
+ ? null
+ : PojoToRowConverter.of(pojoClass, tableSchema, targetProjection);
+ }
+
+ @Override
+ public CompletableFuture<UpsertResult> upsert(T record) {
+ if (record instanceof InternalRow) {
+ return delegate.upsert((InternalRow) record);
+ }
+ InternalRow row = convertPojo(record, /*forDelete=*/ false);
Review Comment:
WriteRecord is not a format conversion; it represents operations
--
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] [Client] Add Pojo Support [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3681500549
@wuchong I made the required changes.
Let me know if this approach resonates and works better
--
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] [Client] Add Pojo Support [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3685265468
Thank you @polyzos , I will take another look.
--
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] [Client] Add Pojo Support [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3576121461
> Can you provide some more context in terms of what you mean - push the conversion further down?
> On the scan side, I see you mentioned optimization in the CompletedFetch.toScanRecord(LogRecord), that is something indeed that I didn't think of and might result in some further optimization.
My aim is so that we can eliminate performance penalty by avoiding performing conversion twice. By "pushing conversion down", I mean you can do something like moving the RowToPojoConverter method calls into LogScannerImpl/CompletedFetch. Strategy pattern could be useful here.
> Is there something also on the write side you are thinking of?
Currently not, I'm very new to this code base but it's certainly worth exploring especially if we're also converting twice on the write side. Happy to have a look at the write side as well.
--
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] [Client] Add Pojo Support [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on code in PR #1992:
URL: https://github.com/apache/fluss/pull/1992#discussion_r2560825627
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedAppendWriter.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link AppendWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedAppendWriter<T> implements AppendWriter<T> {
+
+ private final AppendWriterImpl delegate;
+ private final Class<T> pojoClass;
+ private final RowType tableSchema;
+ private final PojoToRowConverter<T> pojoToRowConverter;
+
+ TypedAppendWriter(AppendWriterImpl delegate, Class<T> pojoClass, TableInfo tableInfo) {
+ this.delegate = delegate;
+ this.pojoClass = pojoClass;
+ this.tableSchema = tableInfo.getRowType();
+ this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
+ }
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ @Override
+ public CompletableFuture<AppendResult> append(T record) {
+ if (record instanceof InternalRow) {
+ return delegate.append((InternalRow) record);
+ }
+ InternalRow row = pojoToRowConverter.toRow(record);
Review Comment:
the main conversion here Pojo -> InternalRow.
IndexedRow is of type InternalRow and is used when the user specifies indexed as the log format.
Arrow is the default one.
--
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] [Client] Add Pojo Support [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong merged PR #1992:
URL: https://github.com/apache/fluss/pull/1992
--
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] [Client] Add Pojo Support [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3545977582
FYI @wuchong .. This PR adds support for writing/scanning Pojos directly with the client, while keeping the API as is.
The only difference is that now it's typed.
--
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] [Client] Add Pojo Support [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #1992:
URL: https://github.com/apache/fluss/pull/1992#discussion_r2559545879
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/TypedLogScanner.java:
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.scanner.log;
+
+import org.apache.fluss.client.converter.RowToPojoConverter;
+import org.apache.fluss.client.table.scanner.ScanRecord;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Adapter that converts {@link InternalRow} records from a {@link LogScanner} into POJOs of type T.
+ */
+public class TypedLogScanner<T> implements LogScanner<T> {
+
+ private final LogScanner<InternalRow> delegate;
+ private final RowToPojoConverter<T> converter;
+
+ public TypedLogScanner(
+ LogScanner<InternalRow> delegate,
+ Class<T> pojoClass,
+ TableInfo tableInfo,
+ int[] projectedColumns) {
+ this.delegate = delegate;
+ RowType tableSchema = tableInfo.getRowType();
+ RowType projection =
+ projectedColumns == null ? tableSchema : tableSchema.project(projectedColumns);
+ this.converter = RowToPojoConverter.of(pojoClass, tableSchema, projection);
+ }
+
+ @Override
+ public ScanRecords<T> poll(Duration timeout) {
+ ScanRecords<InternalRow> records = delegate.poll(timeout);
+ if (records == null || records.isEmpty()) {
+ return ScanRecords.empty();
+ }
+ Map<TableBucket, List<ScanRecord<T>>> out = new HashMap<>();
+ for (TableBucket bucket : records.buckets()) {
+ List<ScanRecord<InternalRow>> list = records.records(bucket);
+ List<ScanRecord<T>> converted = new ArrayList<>(list.size());
+ for (ScanRecord<InternalRow> r : list) {
+ InternalRow row = r.getValue();
+ T pojo = converter.fromRow(row);
+ converted.add(
+ new ScanRecord<>(r.logOffset(), r.timestamp(), r.getChangeType(), pojo));
+ }
+ out.put(bucket, converted);
Review Comment:
I like the clean delegate approach, however what is the performance implication here? We would be looking at 2x memory usage per poll and 2x conversions when compared against non-POJO use cases?
Should POJO conversion be pushed down to e.g. `CompletedFetch.toScanRecord(LogRecord)`? Alternatively, we should note this for future improvement.
##########
fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java:
##########
@@ -73,12 +74,13 @@ static <T> PojoType<T> of(Class<T> pojoClass) {
for (Map.Entry<String, Field> e : allFields.entrySet()) {
String name = e.getKey();
Field field = e.getValue();
+ // Enforce nullable fields: primitives are not allowed in POJO definitions.
if (field.getType().isPrimitive()) {
throw new IllegalArgumentException(
- String.format(
- "POJO class %s has primitive field '%s' of type %s. Primitive types are not allowed; all fields must be nullable (use wrapper types).",
- pojoClass.getName(), name, field.getType().getName()));
+ "Primitive types are not allowed; all fields must be nullable (use wrapper types).");
Review Comment:
Curious on why we're removing information that would be useful for debugging from logs? It would be nice to at least retain pojoClass name so that users know which pojo the IAE is thrown for.
--
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] [Client] Add Pojo Support [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #1992:
URL: https://github.com/apache/fluss/pull/1992#discussion_r2560418511
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedUpsertWriter.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.GenericRow;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link UpsertWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedUpsertWriter<T> implements UpsertWriter<T> {
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ private final UpsertWriterImpl delegate;
+ private final Class<T> pojoClass;
+ private final TableInfo tableInfo;
+ private final RowType tableSchema;
+ private final int[] targetColumns; // may be null
+
+ private final RowType pkProjection;
+ @Nullable private final RowType targetProjection;
+
+ private final PojoToRowConverter<T> pojoToRowConverter;
+ private final PojoToRowConverter<T> pkConverter;
+ @Nullable private final PojoToRowConverter<T> targetConverter;
+
+ TypedUpsertWriter(
+ UpsertWriterImpl delegate,
+ Class<T> pojoClass,
+ TableInfo tableInfo,
+ int[] targetColumns) {
+ this.delegate = delegate;
+ this.pojoClass = pojoClass;
+ this.tableInfo = tableInfo;
+ this.tableSchema = tableInfo.getRowType();
+ this.targetColumns = targetColumns;
+
+ // Precompute projections
+ this.pkProjection = this.tableSchema.project(tableInfo.getPhysicalPrimaryKeys());
+ this.targetProjection =
+ (targetColumns == null) ? null : this.tableSchema.project(targetColumns);
+
+ // Initialize reusable converters
+ this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
+ this.pkConverter = PojoToRowConverter.of(pojoClass, tableSchema, pkProjection);
+ this.targetConverter =
+ (targetProjection == null)
+ ? null
+ : PojoToRowConverter.of(pojoClass, tableSchema, targetProjection);
+ }
+
+ @Override
+ public CompletableFuture<UpsertResult> upsert(T record) {
+ if (record instanceof InternalRow) {
+ return delegate.upsert((InternalRow) record);
+ }
+ InternalRow row = convertPojo(record, /*forDelete=*/ false);
Review Comment:
Similar comment here on avoiding converting twice. (POJO -> InternalRow -> WriteRecord).
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/TypedAppendWriter.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.client.table.writer;
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A typed {@link AppendWriter} that converts POJOs to {@link InternalRow} and delegates to the
+ * existing internal-row based writer implementation.
+ */
+class TypedAppendWriter<T> implements AppendWriter<T> {
+
+ private final AppendWriterImpl delegate;
+ private final Class<T> pojoClass;
+ private final RowType tableSchema;
+ private final PojoToRowConverter<T> pojoToRowConverter;
+
+ TypedAppendWriter(AppendWriterImpl delegate, Class<T> pojoClass, TableInfo tableInfo) {
+ this.delegate = delegate;
+ this.pojoClass = pojoClass;
+ this.tableSchema = tableInfo.getRowType();
+ this.pojoToRowConverter = PojoToRowConverter.of(pojoClass, tableSchema, tableSchema);
+ }
+
+ @Override
+ public void flush() {
+ delegate.flush();
+ }
+
+ @Override
+ public CompletableFuture<AppendResult> append(T record) {
+ if (record instanceof InternalRow) {
+ return delegate.append((InternalRow) record);
+ }
+ InternalRow row = pojoToRowConverter.toRow(record);
Review Comment:
Similar comment here on avoiding converting twice. (POJO -> InternalRow -> IndexedRow).
--
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] [Client] Add Pojo Support [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on code in PR #1992:
URL: https://github.com/apache/fluss/pull/1992#discussion_r2637879234
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/TypedLookuper.java:
##########
@@ -0,0 +1,73 @@
+package org.apache.fluss.client.lookup;
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import org.apache.fluss.client.converter.PojoToRowConverter;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Decorator for {@link Lookuper} that enables generic key lookup via {@link
+ * Lookuper#lookup(Object)}. Converts POJO keys to {@link InternalRow} using existing converters
+ * based on table schema and active lookup columns, and directly delegates when the key is already
+ * an {@link InternalRow}.
+ */
+final class TypedLookuper<K> implements Lookuper<K> {
+
+ private final Lookuper<InternalRow> delegate;
+ private final TableInfo tableInfo;
+ @Nullable private final List<String> lookupColumnNames;
+
+ TypedLookuper(
+ Lookuper<InternalRow> delegate,
+ TableInfo tableInfo,
+ @Nullable List<String> lookupColumnNames) {
+ this.delegate = delegate;
+ this.tableInfo = tableInfo;
+ this.lookupColumnNames = lookupColumnNames;
+ }
+
+ @Override
+ public CompletableFuture<LookupResult> lookup(K key) {
+ if (key == null) {
+ throw new IllegalArgumentException("key must not be null");
+ }
+ // Fast-path: already an InternalRow
+ if (key instanceof InternalRow) {
+ return delegate.lookup((InternalRow) key);
+ }
+ RowType tableSchema = tableInfo.getRowType();
+ RowType keyProjection;
+ if (lookupColumnNames == null) {
+ keyProjection = tableSchema.project(tableInfo.getPrimaryKeys());
+ } else {
+ keyProjection = tableSchema.project(lookupColumnNames);
+ }
+ @SuppressWarnings("unchecked")
+ Class<K> keyClass = (Class<K>) key.getClass();
+ PojoToRowConverter<K> keyConv = PojoToRowConverter.of(keyClass, tableSchema, keyProjection);
Review Comment:
These are very heavy operations and should be invoked on hot-path. The `PojoToRowConverter` should be initialized in the constructor. This can improve the performance a lot.
##########
fluss-client/src/main/java/org/apache/fluss/client/lookup/Lookup.java:
##########
@@ -93,7 +99,11 @@ default Lookup lookupBy(String... lookupColumnNames) {
* lookup columns. By default, the lookup columns are the primary key columns, but can be
* changed with ({@link #lookupBy(List)}) method.
*
+ * <p>The returned lookuper accepts generic keys of type {@code K}. If a key is a POJO, the
+ * client implementation will convert it to an {@code InternalRow} based on the table schema and
+ * the active lookup columns.
+ *
* @return the lookuper
*/
- Lookuper createLookuper();
+ <K> Lookuper<K> createLookuper();
Review Comment:
This is a big change for Fluss users, as `Lookuper`, `Appender`, and `Upserter` are fundamental data access interfaces. To maintain backward compatibility, we should **keep these existing interfaces unchanged**.
Instead, we can introduce a new generic interface: `TypedLookuper<T>`. This interface adds a type-safe lookup method:
```java
CompletableFuture<LookupResult> lookup(T lookupKey);
```
We can then provide a new factory method in `Lookup` to create it:
```java
<K> TypedLookuper<T> createTypedLookuper(Class<T> pojoClass);
```
This approach preserves existing APIs while enabling type-safe lookups for users who need them.
##########
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/Scan.java:
##########
@@ -63,7 +64,14 @@ public interface Scan {
*
* <p>Note: this API doesn't support pre-configured with {@link #limit(int)}.
*/
- LogScanner createLogScanner();
+ LogScanner<InternalRow> createLogScanner();
+
+ /**
+ * Creates a typed LogScanner to continuously read log data as POJOs of the given class.
+ *
+ * <p>Note: this API doesn't support pre-configured with {@link #limit(int)}.
+ */
+ <T> LogScanner<T> createLogScanner(Class<T> pojoClass);
Review Comment:
Same to `TypedLookuper`, we should keep `LogScanner` unchanged and introduce
```java
public interface TypedLogScanner<T> {
@Override
TypedScanRecords<T> poll(Duration timeout);
}
public class TypedScanRecords<T> extends ScanRecords {
...
@Override
public Iterator<TypedScanRecord> iterator() {
...
}
}
public class TypedScanRecord<T> extends ScanRecord {
private final T pojo;
}
```
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/Upsert.java:
##########
@@ -60,8 +57,11 @@ public interface Upsert {
Upsert partialUpdate(String... targetColumnNames);
/**
- * Create a new {@link UpsertWriter} with the optional {@link #partialUpdate(String...)}
- * information to upsert and delete data to a Primary Key Table.
+ * Create a new {@link UpsertWriter} using {@code InternalRow} with the optional {@link
+ * #partialUpdate(String...)} information to upsert and delete data to a Primary Key Table.
*/
- UpsertWriter createWriter();
+ UpsertWriter<?> createWriter();
+
+ /** Create a new typed {@link UpsertWriter} to write POJOs directly. */
+ <T> UpsertWriter<T> createWriter(Class<T> pojoClass);
Review Comment:
ditto
##########
fluss-client/src/main/java/org/apache/fluss/client/table/writer/Append.java:
##########
@@ -32,6 +32,9 @@ public interface Append {
// TODO: Add more methods to configure the AppendWriter, such as apply static partitions,
// apply overwrites, etc.
- /** Create a new {@link AppendWriter} to write data to a Log Table. */
- AppendWriter createWriter();
+ /** Create a new {@link AppendWriter} to write data to a Log Table using InternalRow. */
+ AppendWriter<?> createWriter();
+
+ /** Create a new typed {@link AppendWriter} to write POJOs directly. */
+ <T> AppendWriter<T> createWriter(Class<T> pojoClass);
Review Comment:
ditto
--
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] [Client] Add Pojo Support [fluss]
Posted by "wuchong (via GitHub)" <gi...@apache.org>.
wuchong commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3680260468
@polyzos Yeah, I noticed those typed classes. However, typed classes are internal, only **interfaces are visible to users**.
My suggestion is to introduce **dedicated typed interfaces** (e.g., `interface TypedLookuper<T>`) that are **separate from the existing `Lookuper` interface**, rather than making `Lookuper` itself generic.
While turning `Lookuper` into a generic interface would be *binary-compatible*, it would introduce **type erasure warnings in IDEs** and clutter the public API. More importantly, for use cases involving `InternalRow`, the generic type `T` has **no semantic meaning**, so forcing a type parameter there adds unnecessary noise without benefit.
By keeping `Lookuper` and `TypedLookuper<T>` as distinct interfaces, we maintain clean separation of concerns:
- `Lookuper` for low-level, type-agnostic access (e.g., `InternalRow`)
- `TypedLookuper<T>` for high-level, type-safe lookups
This approach preserves backward compatibility, avoids IDE warnings, and aligns with how users actually interact with the API.
--
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] [Client] Add Pojo Support [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3576057785
@leekeiabstraction yes, it's a separate thread.
Can you provide some more context in terms of what you mean - push the conversion further down?
On the scan side, I see you mentioned optimization in the `CompletedFetch.toScanRecord(LogRecord)`, that is something indeed that I didn't think of and might result in some further optimization.
Is there something also on the write side you are thinking of?
--
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] [Client] Add Pojo Support [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on PR #1992:
URL: https://github.com/apache/fluss/pull/1992#issuecomment-3576001847
Thank you @polyzos for addressing the commends and also providing the data from your tests! That gives a clear picture on the performance. Further response questions below:
> This is basically a trade-off for the users. Using InternalRow/GenericRow directly is way more efficient; however, this might come with some extra complexity and boilerplatete code.
> For this reason, I want to give flexibility, probably leave the docs as is, with GenericRow being the go-to approach, but also add a section that Pojos can be used directly and maybe highlight this trade-off.
IMO, this does not need to be a trade-off. I am curious if you have explored implementation complexity or cons around pushing the Pojo conversion down further so that conversion to/from `InternalRow` can be skipped altogether? From a quick look, it does seem like you have most of the interfaces updated to support it.
> Moving forward I'm thinking that maybe it makes sense to add some helper classes that also derive the schema for the table from a Pojo.
Trying to understand, does this relate to performance or a separate thread on further changes that you're planning?
--
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