refactor to typed apis

Author: polyzos

fluss-client/src/main/java/org/apache/fluss/client/lookup/Lookup.java

@@ -36,7 +36,7 @@
  * column [k STRING]:
  *
  * <pre>{@code
- * Lookuper<InternalRow> lookuper = table.newLookup().createLookuper();
+ * Lookuper lookuper = table.newLookup().createLookuper();
  * CompletableFuture<LookupResult> resultFuture = lookuper.lookup(GenericRow.of("key1"));
  * resultFuture.get().getRowList().forEach(System.out::println);
  * }</pre>
@@ -45,15 +45,15 @@
  * columns [a INT, b STRING, c BIGINT] and bucket key [a, b]:
  *
  * <pre>{@code
- * Lookuper<InternalRow> lookuper = table.newLookup().lookupBy("a", "b").createLookuper();
+ * Lookuper lookuper = table.newLookup().lookupBy("a", "b").createLookuper();
  * CompletableFuture<LookupResult> resultFuture = lookuper.lookup(GenericRow.of(1, "b1"));
  * resultFuture.get().getRowList().forEach(System.out::println);
  * }</pre>
  *
  * <p>Example 3: Using a POJO key (conversion handled internally):
  *
  * <pre>{@code
- * Lookuper<MyKeyPojo> lookuper = table.newLookup().createLookuper();
+ * TypedLookuper<MyKeyPojo> lookuper = table.newLookup().createTypedLookuper(MyKeyPojo.class);
  * LookupResult result = lookuper.lookup(new MyKeyPojo(...)).get();
  * }</pre>
  *
@@ -99,11 +99,16 @@ 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
      */
-    <K> Lookuper<K> createLookuper();
+    Lookuper createLookuper();
+
+    /**
+     * Creates a {@link TypedLookuper} instance to lookup rows of a primary key table using POJOs.
+     *
+     * @param pojoClass the class of the POJO
+     * @param <T> the type of the POJO
+     * @return the typed lookuper
+     */
+    <T> TypedLookuper<T> createTypedLookuper(Class<T> pojoClass);
 }

fluss-client/src/main/java/org/apache/fluss/client/lookup/Lookuper.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.client.lookup;
 
 import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.row.InternalRow;
 
 import javax.annotation.concurrent.NotThreadSafe;
 
@@ -27,30 +28,19 @@
  * A lookuper performs key-based lookups against a primary key table, using either the full primary
  * key or a prefix of the primary key (when configured via {@code Lookup#lookupBy}).
  *
- * <p>This interface is generic on the key type {@code K}: - When used in row mode, implementations
- * are typically declared as {@code Lookuper<InternalRow>} and accept an {@link
- * org.apache.fluss.row.InternalRow} containing the key fields in the configured order. - When used
- * with POJO keys, client-provided lookuper instances can also accept a POJO key type (for example
- * {@code Lookuper<MyKeyPojo>}) and will transparently convert the POJO to an {@code InternalRow}
- * using the table schema and active lookup columns.
- *
  * <p>Usage examples:
  *
  * <pre>{@code
  * // Row-based key (InternalRow)
- * Lookuper<InternalRow> lookuper = table.newLookup().createLookuper();
+ * Lookuper lookuper = table.newLookup().createLookuper();
  * LookupResult res = lookuper.lookup(keyRow).get();
- *
- * // POJO key (converted internally)
- * Lookuper<MyKeyPojo> lookuperPojo = table.newLookup().createLookuper();
- * LookupResult res2 = lookuperPojo.lookup(new MyKeyPojo(...)).get();
  * }</pre>
  *
  * @since 0.6
  */
 @PublicEvolving
 @NotThreadSafe
-public interface Lookuper<K> {
+public interface Lookuper {
 
     /**
      * Lookups certain row from the given lookup key.
@@ -59,12 +49,8 @@ public interface Lookuper<K> {
      * {@code table.newLookup().createLookuper()}), or be the prefix key if the lookuper is a Prefix
      * Key Lookuper (created by {@code table.newLookup().lookupBy(prefixKeys).createLookuper()}).
      *
-     * <p>The key can be either an {@link org.apache.fluss.row.InternalRow} or a POJO representing
-     * the lookup key. Client-provided implementations returned by the Fluss client handle POJO-to-
-     * row conversion internally when necessary.
-     *
      * @param lookupKey the lookup key
      * @return the result of lookup.
      */
-    CompletableFuture<LookupResult> lookup(K lookupKey);
+    CompletableFuture<LookupResult> lookup(InternalRow lookupKey);
 }

fluss-client/src/main/java/org/apache/fluss/client/lookup/PrefixKeyLookuper.java

@@ -46,13 +46,7 @@
  * of the primary key.
  */
 @NotThreadSafe
-class PrefixKeyLookuper implements Lookuper<InternalRow> {
-
-    private final TableInfo tableInfo;
-
-    private final MetadataUpdater metadataUpdater;
-
-    private final LookupClient lookupClient;
+class PrefixKeyLookuper extends AbstractLookuper implements Lookuper {
 
     /** Extract bucket key from prefix lookup key row. */
     private final KeyEncoder bucketKeyEncoder;

fluss-client/src/main/java/org/apache/fluss/client/lookup/PrimaryKeyLookuper.java

@@ -40,13 +40,7 @@
 
 /** An implementation of {@link Lookuper} that lookups by primary key. */
 @NotThreadSafe
-class PrimaryKeyLookuper implements Lookuper<InternalRow> {
-
-    private final TableInfo tableInfo;
-
-    private final MetadataUpdater metadataUpdater;
-
-    private final LookupClient lookupClient;
+class PrimaryKeyLookuper extends AbstractLookuper implements Lookuper {
 
     private final KeyEncoder primaryKeyEncoder;
 

fluss-client/src/main/java/org/apache/fluss/client/lookup/TableLookup.java

@@ -20,7 +20,6 @@
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.metadata.SchemaGetter;
 import org.apache.fluss.metadata.TableInfo;
-import org.apache.fluss.row.InternalRow;
 
 import javax.annotation.Nullable;
 
@@ -64,14 +63,17 @@ public Lookup lookupBy(List<String> lookupColumnNames) {
     }
 
     @Override
-    public <K> Lookuper<K> createLookuper() {
-        Lookuper<InternalRow> lookuper;
+    public Lookuper createLookuper() {
         if (lookupColumnNames == null) {
             return new PrimaryKeyLookuper(tableInfo, schemaGetter, metadataUpdater, lookupClient);
         } else {
             return new PrefixKeyLookuper(
                     tableInfo, schemaGetter, metadataUpdater, lookupClient, lookupColumnNames);
         }
-        return new TypedLookuper<K>(lookuper, tableInfo, lookupColumnNames);
     }
-}
+
+    @Override
+    public <T> TypedLookuper<T> createTypedLookuper(Class<T> pojoClass) {
+        return new TypedLookuperImpl<>(createLookuper(), tableInfo, lookupColumnNames);
+    }
+}

fluss-client/src/main/java/org/apache/fluss/client/lookup/TypedLookuper.java

@@ -1,5 +1,3 @@
-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
@@ -17,57 +15,26 @@
  * 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;
+package org.apache.fluss.client.lookup;
 
-import javax.annotation.Nullable;
+import org.apache.fluss.annotation.PublicEvolving;
 
-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}.
+ * A typed lookuper performs key-based lookups against a primary key table using POJOs.
+ *
+ * @param <T> the type of the lookup key
+ * @since 0.6
  */
-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);
-        InternalRow keyRow = keyConv.toRow(key);
-        return delegate.lookup(keyRow);
-    }
+@PublicEvolving
+public interface TypedLookuper<T> {
+
+    /**
+     * Lookups certain row from the given lookup key.
+     *
+     * @param lookupKey the lookup key
+     * @return the result of lookup.
+     */
+    CompletableFuture<LookupResult> lookup(T lookupKey);
 }

fluss-client/src/main/java/org/apache/fluss/client/lookup/TypedLookuperImpl.java

@@ -0,0 +1,71 @@
+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
+ * TypedLookuper#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 TypedLookuperImpl<K> implements TypedLookuper<K> {
+
+    private final Lookuper delegate;
+    private final TableInfo tableInfo;
+    @Nullable private final List<String> lookupColumnNames;
+
+    TypedLookuperImpl(
+            Lookuper 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);
+        InternalRow keyRow = keyConv.toRow(key);
+        return delegate.lookup(keyRow);
+    }
+}

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/Scan.java

@@ -20,8 +20,8 @@
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.client.table.scanner.batch.BatchScanner;
 import org.apache.fluss.client.table.scanner.log.LogScanner;
+import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
 import org.apache.fluss.metadata.TableBucket;
-import org.apache.fluss.row.InternalRow;
 
 import javax.annotation.Nullable;
 
@@ -64,14 +64,14 @@ public interface Scan {
      *
      * <p>Note: this API doesn't support pre-configured with {@link #limit(int)}.
      */
-    LogScanner<InternalRow> createLogScanner();
+    LogScanner createLogScanner();
 
     /**
-     * Creates a typed LogScanner to continuously read log data as POJOs of the given class.
+     * Creates a {@link TypedLogScanner} 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);
+    <T> TypedLogScanner<T> createTypedLogScanner(Class<T> pojoClass);
 
     /**
      * Creates a {@link BatchScanner} to read current data in the given table bucket for this scan.

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java

@@ -26,12 +26,12 @@
 import org.apache.fluss.client.table.scanner.log.LogScanner;
 import org.apache.fluss.client.table.scanner.log.LogScannerImpl;
 import org.apache.fluss.client.table.scanner.log.TypedLogScanner;
+import org.apache.fluss.client.table.scanner.log.TypedLogScannerImpl;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.exception.FlussRuntimeException;
 import org.apache.fluss.metadata.SchemaGetter;
 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 javax.annotation.Nullable;
@@ -116,9 +116,9 @@ public LogScanner createLogScanner() {
     }
 
     @Override
-    public <T> LogScanner<T> createLogScanner(Class<T> pojoClass) {
-        LogScanner<InternalRow> base = createLogScanner();
-        return new TypedLogScanner<>(base, pojoClass, tableInfo, projectedColumns);
+    public <T> TypedLogScanner<T> createTypedLogScanner(Class<T> pojoClass) {
+        LogScanner base = createLogScanner();
+        return new TypedLogScannerImpl<>(base, pojoClass, tableInfo, projectedColumns);
     }
 
     @Override

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScanner.java

@@ -18,6 +18,7 @@
 package org.apache.fluss.client.table.scanner.log;
 
 import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.row.InternalRow;
 
 import java.time.Duration;
 
@@ -27,7 +28,7 @@
  * @since 0.1
  */
 @PublicEvolving
-public interface LogScanner<T> extends AutoCloseable {
+public interface LogScanner extends AutoCloseable {
 
     /**
      * The earliest offset to fetch from. Fluss uses "-2" to indicate fetching from log start
@@ -48,7 +49,7 @@ public interface LogScanner<T> extends AutoCloseable {
      * @throws java.lang.IllegalStateException if the scanner is not subscribed to any buckets to
      *     read from.
      */
-    ScanRecords<T> poll(Duration timeout);
+    ScanRecords<InternalRow> poll(Duration timeout);
 
     /**
      * Subscribe to the given table bucket in given offset dynamically. If the table bucket is