Merge main branch into fix-2128-datalake-add-column-mvp

- Merged latest main changes including Array type support
- Resolved test conflicts in FlussRecordAsPaimonRowTest.java
- Retained both Array tests and schema evolution tests

Author: buvb

.github/workflows/ci-template.yaml

@@ -36,7 +36,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        module: [ core, flink, lake ]
+        module: [ core, flink, spark3, lake ]
     name: "${{ matrix.module }}"
     steps:
       - name: Checkout code

.github/workflows/stage.sh

@@ -19,15 +19,30 @@
 
 STAGE_CORE="core"
 STAGE_FLINK="flink"
+STAGE_SPARK="spark3"
 STAGE_LAKE="lake"
 
 MODULES_FLINK="\
 fluss-flink,\
 fluss-flink/fluss-flink-common,\
-fluss-flink/fluss-flink-2.1,\
+fluss-flink/fluss-flink-2.2,\
 fluss-flink/fluss-flink-1.20,\
 "
 
+MODULES_COMMON_SPARK="\
+fluss-spark,\
+fluss-spark/fluss-spark-common,\
+fluss-spark/fluss-spark-ut,\
+"
+
+MODULES_SPARK3="\
+fluss-spark,\
+fluss-spark/fluss-spark-common,\
+fluss-spark/fluss-spark-ut,\
+fluss-spark/fluss-spark-3.5,\
+fluss-spark/fluss-spark-3.4,\
+"
+
 # we move Flink legacy version tests to "lake" module for balancing testing time
 MODULES_LAKE="\
 fluss-flink/fluss-flink-1.19,\
@@ -42,10 +57,12 @@ function get_test_modules_for_stage() {
     local stage=$1
 
     local modules_flink=$MODULES_FLINK
+    local modules_spark3=$MODULES_SPARK3
     local modules_lake=$MODULES_LAKE
     local negated_flink=\!${MODULES_FLINK//,/,\!}
+    local negated_spark=\!${MODULES_COMMON_SPARK//,/,\!}
     local negated_lake=\!${MODULES_LAKE//,/,\!}
-    local modules_core="$negated_flink,$negated_lake"
+    local modules_core="$negated_flink,$negated_spark,$negated_lake"
 
     case ${stage} in
         (${STAGE_CORE})
@@ -54,6 +71,9 @@ function get_test_modules_for_stage() {
         (${STAGE_FLINK})
             echo "-pl fluss-test-coverage,$modules_flink"
         ;;
+        (${STAGE_SPARK})
+            echo "-Pspark3 -pl fluss-test-coverage,$modules_spark3"
+        ;;
         (${STAGE_LAKE})
             echo "-pl fluss-test-coverage,$modules_lake"
         ;;

.scalafmt.conf

@@ -0,0 +1,70 @@
+runner.dialect = scala212
+
+# Version is required to make sure IntelliJ picks the right version
+version = 3.10.2
+preset = default
+
+# Max column
+maxColumn = 100
+
+# This parameter simply says the .stripMargin method was not redefined by the user to assign
+# special meaning to indentation preceding the | character. Hence, that indentation can be modified.
+assumeStandardLibraryStripMargin = true
+align.stripMargin = true
+
+# Align settings
+align.preset = none
+align.closeParenSite = false
+align.openParenCallSite = false
+danglingParentheses.defnSite = false
+danglingParentheses.callSite = false
+danglingParentheses.ctrlSite = true
+danglingParentheses.tupleSite = false
+align.openParenCallSite = false
+align.openParenDefnSite = false
+align.openParenTupleSite = false
+
+# Newlines
+newlines.alwaysBeforeElseAfterCurlyIf = false
+newlines.beforeCurlyLambdaParams = multiline # Newline before lambda params
+newlines.afterCurlyLambdaParams = squash # No newline after lambda params
+newlines.inInterpolation = "avoid"
+newlines.avoidInResultType = true
+optIn.annotationNewlines = true
+
+# Scaladoc
+docstrings.style = Asterisk # Javadoc style
+docstrings.removeEmpty = true
+docstrings.oneline = fold
+docstrings.forceBlankLineBefore = true
+
+# Indentation
+indent.extendSite = 2 # This makes sure extend is not indented as the ctor parameters
+
+# Rewrites
+rewrite.rules = [AvoidInfix, Imports, RedundantBraces, SortModifiers]
+
+# Imports
+rewrite.imports.sort = scalastyle
+rewrite.imports.groups = [
+    ["org.apache.fluss\\..*"],
+    ["org.apache.fluss.shade\\..*"],
+    [".*"],
+    ["javax\\..*"],
+    ["java\\..*"],
+    ["scala\\..*"]
+]
+rewrite.imports.contiguousGroups = no
+importSelectors = singleline # Imports in a single line, like IntelliJ
+
+# Remove redundant braces in string interpolation.
+rewrite.redundantBraces.stringInterpolation = true
+rewrite.redundantBraces.defnBodies = false
+rewrite.redundantBraces.generalExpressions = false
+rewrite.redundantBraces.ifElseExpressions = false
+rewrite.redundantBraces.methodBodies = false
+rewrite.redundantBraces.includeUnitMethods = false
+rewrite.redundantBraces.maxBreaks = 1
+
+# Remove trailing commas
+rewrite.trailingCommas.style = "never"

…i/connector/sink2/WriterInitContext.java copyright.txtfluss-flink/fluss-flink-1.18/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContext.java renamed to copyright.txt fluss-flink/fluss-flink-1.18/src/main/java/org/apache/flink/api/connector/sink2/WriterInitContext.java renamed to copyright.txt

@@ -15,7 +15,3 @@
  * limitations under the License.
  */
 
-package org.apache.flink.api.connector.sink2;
-
-/** Placeholder class to resolve compatibility issues. */
-public interface WriterInitContext extends Sink.InitContext {}

fluss-client/src/main/java/org/apache/fluss/client/converter/ConverterCommons.java

@@ -83,6 +83,24 @@ static void validatePojoMatchesTable(PojoType<?> pojoType, RowType tableSchema)
         }
     }
 
+    static void validatePojoMatchesProjection(PojoType<?> pojoType, RowType projection) {
+        Set<String> pojoNames = pojoType.getProperties().keySet();
+        List<String> fieldNames = projection.getFieldNames();
+        if (!pojoNames.containsAll(fieldNames)) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "POJO fields %s must contain all projection fields %s. "
+                                    + "For full-table writes, POJO fields must exactly match table schema fields.",
+                            pojoNames, fieldNames));
+        }
+        for (int i = 0; i < projection.getFieldCount(); i++) {
+            String name = fieldNames.get(i);
+            DataType dt = projection.getTypeAt(i);
+            PojoType.Property prop = pojoType.getProperty(name);
+            validateCompatibility(dt, prop);
+        }
+    }
+
     static void validateProjectionSubset(RowType projection, RowType tableSchema) {
         Set<String> tableNames = new HashSet<>(tableSchema.getFieldNames());
         for (String n : projection.getFieldNames()) {

fluss-client/src/main/java/org/apache/fluss/client/converter/PojoToRowConverter.java

@@ -30,6 +30,7 @@
 import javax.annotation.Nullable;
 
 import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -54,7 +55,9 @@ private PojoToRowConverter(PojoType<T> pojoType, RowType tableSchema, RowType pr
         this.tableSchema = tableSchema;
         this.projection = projection;
         this.projectionFieldNames = projection.getFieldNames();
-        ConverterCommons.validatePojoMatchesTable(pojoType, tableSchema);
+        // For writer path, allow POJO to be a superset of the projection. It must contain all
+        // projected fields.
+        ConverterCommons.validatePojoMatchesProjection(pojoType, projection);
         ConverterCommons.validateProjectionSubset(projection, tableSchema);
         this.fieldConverters = createFieldConverters();
     }
@@ -177,8 +180,22 @@ private static FieldToRow createFieldConverter(PojoType.Property prop, DataType
                     String.format(
                             "Field %s is not a BigDecimal. Cannot convert to Decimal.", prop.name));
         }
-        return Decimal.fromBigDecimal(
-                (BigDecimal) v, decimalType.getPrecision(), decimalType.getScale());
+        final int precision = decimalType.getPrecision();
+        final int scale = decimalType.getScale();
+
+        // Scale with a deterministic rounding mode to avoid ArithmeticException when rounding is
+        // needed.
+        BigDecimal bd = (BigDecimal) v;
+        BigDecimal scaled = bd.setScale(scale, RoundingMode.HALF_UP);
+
+        if (scaled.precision() > precision) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Decimal value for field %s exceeds precision %d after scaling to %d: %s",
+                            prop.name, precision, scale, scaled));
+        }
+
+        return Decimal.fromBigDecimal(scaled, precision, scale);
     }
 
     /** Converts a LocalDate POJO property to number of days since epoch. */

fluss-client/src/main/java/org/apache/fluss/client/converter/PojoType.java

@@ -37,6 +37,7 @@ final class PojoType<T> {
     private final Class<T> pojoClass;
     private final Constructor<T> defaultConstructor;
     private final Map<String, Property> properties; // property name -> property
+    private static final Map<Class<?>, Class<?>> PRIMITIVE_TO_BOXED = createPrimitiveToBoxedMap();
 
     private PojoType(Class<T> pojoClass, Constructor<T> ctor, Map<String, Property> props) {
         this.pojoClass = pojoClass;
@@ -73,12 +74,15 @@ 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()));
             }
+            // use boxed type as effective type
+            Class<?> effectiveType = boxIfPrimitive(field.getType());
             boolean publicField = Modifier.isPublic(field.getModifiers());
             Method getter = getters.get(name);
             Method setter = setters.get(name);
@@ -94,8 +98,7 @@ static <T> PojoType<T> of(Class<T> pojoClass) {
             }
             props.put(
                     name,
-                    new Property(
-                            name, field.getType(), publicField ? field : null, getter, setter));
+                    new Property(name, effectiveType, publicField ? field : null, getter, setter));
         }
 
         return new PojoType<>(pojoClass, ctor, props);
@@ -235,6 +238,29 @@ private static String decapitalize(String s) {
         return s.substring(0, 1).toLowerCase(Locale.ROOT) + s.substring(1);
     }
 
+    private static Map<Class<?>, Class<?>> createPrimitiveToBoxedMap() {
+        Map<Class<?>, Class<?>> map = new HashMap<>();
+        map.put(boolean.class, Boolean.class);
+        map.put(byte.class, Byte.class);
+        map.put(short.class, Short.class);
+        map.put(int.class, Integer.class);
+        map.put(long.class, Long.class);
+        map.put(float.class, Float.class);
+        map.put(double.class, Double.class);
+        map.put(char.class, Character.class);
+        // void shouldn't appear as a field type, but handle defensively
+        map.put(void.class, Void.class);
+        return map;
+    }
+
+    private static Class<?> boxIfPrimitive(Class<?> type) {
+        if (!type.isPrimitive()) {
+            return type;
+        }
+        Class<?> boxed = PRIMITIVE_TO_BOXED.get(type);
+        return boxed != null ? boxed : type;
+    }
+
     static final class Property {
         final String name;
         final Class<?> type;

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

@@ -24,31 +24,37 @@
 
 /**
  * Used to configure and create a {@link Lookuper} to lookup rows of a primary key table. The built
- * Lookuper can be a primary key lookuper that lookups by the primary key, or a prefix key lookup
- * that lookups by the prefix key of the primary key.
+ * lookuper can lookup by the full primary key, or by a prefix of the primary key when configured
+ * via {@link #lookupBy}.
  *
  * <p>{@link Lookup} objects are immutable and can be shared between threads. Refinement methods,
- * like {@link #lookupBy}, create new Lookup instances.
+ * like {@link #lookupBy}, create new {@code Lookup} instances.
  *
- * <p>Example1: Create a Primary Key Lookuper. Given a table with primary key column [k STRING].
+ * <p>Examples
+ *
+ * <p>Example 1: Primary Key Lookuper using an InternalRow key. Given a table with primary key
+ * column [k STRING]:
  *
  * <pre>{@code
  * Lookuper lookuper = table.newLookup().createLookuper();
  * CompletableFuture<LookupResult> resultFuture = lookuper.lookup(GenericRow.of("key1"));
- * resultFuture.get().getRows().forEach(row -> {
- *    System.out.println(row);
- * });
+ * resultFuture.get().getRowList().forEach(System.out::println);
  * }</pre>
  *
- * <p>Example2: Create a Prefix Key Lookuper. Given a table with primary key column [a INT, b
- * STRING, c BIGINT] and bucket key [a, b].
+ * <p>Example 2: Prefix Key Lookuper using an InternalRow key. Given a table with primary key
+ * columns [a INT, b STRING, c BIGINT] and bucket key [a, b]:
  *
  * <pre>{@code
  * Lookuper lookuper = table.newLookup().lookupBy("a", "b").createLookuper();
  * CompletableFuture<LookupResult> resultFuture = lookuper.lookup(GenericRow.of(1, "b1"));
- * resultFuture.get().getRows().forEach(row -> {
- *   System.out.println(row);
- * });
+ * resultFuture.get().getRowList().forEach(System.out::println);
+ * }</pre>
+ *
+ * <p>Example 3: Using a POJO key (conversion handled internally):
+ *
+ * <pre>{@code
+ * TypedLookuper<MyKeyPojo> lookuper = table.newLookup().createTypedLookuper(MyKeyPojo.class);
+ * LookupResult result = lookuper.lookup(new MyKeyPojo(...)).get();
  * }</pre>
  *
  * @since 0.6
@@ -96,4 +102,13 @@ default Lookup lookupBy(String... lookupColumnNames) {
      * @return the lookuper
      */
     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/LookupResult.java

@@ -27,7 +27,9 @@
 import java.util.Objects;
 
 /**
- * The result of {@link Lookuper#lookup(InternalRow)}.
+ * The result of a lookup operation performed by a {@link Lookuper}. It carries zero, one, or many
+ * {@link org.apache.fluss.row.InternalRow} values depending on whether the underlying lookup is a
+ * primary-key lookup (at most one) or a prefix-key lookup (zero or more).
  *
  * @since 0.1
  */

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

@@ -25,11 +25,16 @@
 import java.util.concurrent.CompletableFuture;
 
 /**
- * The lookup-er is used to lookup row of a primary key table by primary key or prefix key. The
- * lookuper has retriable ability to handle transient errors during lookup operations which is
- * configured by {@link org.apache.fluss.config.ConfigOptions#CLIENT_LOOKUP_MAX_RETRIES}.
+ * 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>Note: Lookuper instances are not thread-safe.
+ * <p>Usage examples:
+ *
+ * <pre>{@code
+ * // Row-based key (InternalRow)
+ * Lookuper lookuper = table.newLookup().createLookuper();
+ * LookupResult res = lookuper.lookup(keyRow).get();
+ * }</pre>
  *
  * @since 0.6
  */
@@ -44,7 +49,7 @@ public interface Lookuper {
      * {@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()}).
      *
-     * @param lookupKey the lookup key.
+     * @param lookupKey the lookup key
      * @return the result of lookup.
      */
     CompletableFuture<LookupResult> lookup(InternalRow lookupKey);