Merge branch 'main' into numeric-type-widening-converters

Author: hemanthsavasere

.github/ISSUE_TEMPLATE/bug.yml

@@ -39,9 +39,8 @@ body:
       description: What Fluss version are you using?
       multiple: false
       options:
-        - "0.7.0 (latest release)"
-        - "0.6.0"
-        - "0.5.0"
+        - "0.8.0 (latest release)"
+        - "0.7.0"
         - "main (development)"
     validations:
       required: true

LICENSE

@@ -401,3 +401,7 @@ LightProto
 ./fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufMessage.java
 ./fluss-protogen/fluss-protogen-generator/src/main/java/org/apache/fluss/protogen/generator/generator/ProtobufNumberField.java
 ./fluss-protogen/fluss-protogen-maven-plugin/src/main/java/org/apache/fluss/protogen/maven/plugin/ProtoGenMojo.java
+
+Apache Maven Wrapper
+./mvnw
+./mvnw.cmd

NOTICE

@@ -21,4 +21,19 @@ with the following copyright notice:
 LightProto
 Copyright 2020 Splunk Inc.
 
-----------------------------------------------------------
+----------------------------------------------------------
+
+This product contains code from the Apache Maven Wrapper Project:
+
+Apache Maven Wrapper
+Copyright 2013-2025 The Apache Software Foundation
+
+This product includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+The original idea and initial implementation of the maven-wrapper module is derived
+from the Gradle Wrapper which was written originally by Hans Dockter and Adam Murdoch.
+Copyright 2007 the original author or authors.
+
+----------------------------------------------------------
+

README.md

@@ -1,3 +1,21 @@
+<!--
+ 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.
+-->
+
 <p align="center">
     <picture>
       <source media="(prefers-color-scheme: dark)" srcset="website/static/img/logo/svg/white_color_logo.svg">

docker/quickstart-flink/README.md

@@ -1,3 +1,21 @@
+<!--
+ 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.
+-->
+
 # Fluss Quickstart Flink Docker
 
 This directory contains the Docker setup for Fluss Quickstart with Flink integration.

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

@@ -24,8 +24,9 @@
 
 import java.math.BigDecimal;
 import java.util.Arrays;
-import java.util.HashMap;
+import java.util.EnumMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
@@ -43,7 +44,7 @@ final class ConverterCommons {
     static final Map<DataTypeRoot, Set<Class<?>>> SUPPORTED_TYPES = createSupportedTypes();
 
     private static Map<DataTypeRoot, Set<Class<?>>> createSupportedTypes() {
-        Map<DataTypeRoot, Set<Class<?>>> map = new HashMap<>();
+        Map<DataTypeRoot, Set<Class<?>>> map = new EnumMap<>(DataTypeRoot.class);
         map.put(DataTypeRoot.BOOLEAN, setOf(Boolean.class));
 
         // Numeric types with widening support
@@ -94,15 +95,15 @@ private static Map<DataTypeRoot, Set<Class<?>>> createSupportedTypes() {
 
     static void validatePojoMatchesTable(PojoType<?> pojoType, RowType tableSchema) {
         Set<String> pojoNames = pojoType.getProperties().keySet();
-        Set<String> tableNames = new HashSet<>(tableSchema.getFieldNames());
-        if (!pojoNames.equals(tableNames)) {
+        List<String> fieldNames = tableSchema.getFieldNames();
+        if (!pojoNames.containsAll(fieldNames)) {
             throw new IllegalArgumentException(
                     String.format(
                             "POJO fields %s must exactly match table schema fields %s.",
-                            pojoNames, tableNames));
+                            pojoNames, fieldNames));
         }
         for (int i = 0; i < tableSchema.getFieldCount(); i++) {
-            String name = tableSchema.getFieldNames().get(i);
+            String name = fieldNames.get(i);
             DataType dt = tableSchema.getTypeAt(i);
             PojoType.Property prop = pojoType.getProperty(name);
             validateCompatibility(dt, prop);

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

@@ -23,6 +23,7 @@
 import org.apache.fluss.row.TimestampLtz;
 import org.apache.fluss.row.TimestampNtz;
 import org.apache.fluss.types.DataType;
+import org.apache.fluss.types.DataTypeChecks;
 import org.apache.fluss.types.DecimalType;
 import org.apache.fluss.types.RowType;
 
@@ -142,9 +143,15 @@ private static FieldToRow createFieldConverter(PojoType.Property prop, DataType
             case TIME_WITHOUT_TIME_ZONE:
                 return (obj) -> convertTimeValue(prop, prop.read(obj));
             case TIMESTAMP_WITHOUT_TIME_ZONE:
-                return (obj) -> convertTimestampNtzValue(prop, prop.read(obj));
+                {
+                    final int precision = DataTypeChecks.getPrecision(fieldType);
+                    return (obj) -> convertTimestampNtzValue(precision, prop, prop.read(obj));
+                }
             case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
-                return (obj) -> convertTimestampLtzValue(prop, prop.read(obj));
+                {
+                    final int precision = DataTypeChecks.getPrecision(fieldType);
+                    return (obj) -> convertTimestampLtzValue(precision, prop, prop.read(obj));
+                }
             default:
                 throw new UnsupportedOperationException(
                         String.format(
@@ -210,9 +217,17 @@ private static FieldToRow createFieldConverter(PojoType.Property prop, DataType
         return (int) (t.toNanoOfDay() / 1_000_000);
     }
 
-    /** Converts a LocalDateTime POJO property to Fluss TimestampNtz. */
+    /**
+     * Converts a LocalDateTime POJO property to Fluss TimestampNtz, respecting the specified
+     * precision.
+     *
+     * @param precision the timestamp precision (0-9)
+     * @param prop the POJO property metadata
+     * @param v the value to convert
+     * @return TimestampNtz with precision applied, or null if v is null
+     */
     private static @Nullable TimestampNtz convertTimestampNtzValue(
-            PojoType.Property prop, @Nullable Object v) {
+            int precision, PojoType.Property prop, @Nullable Object v) {
         if (v == null) {
             return null;
         }
@@ -222,24 +237,82 @@ private static FieldToRow createFieldConverter(PojoType.Property prop, DataType
                             "Field %s is not a LocalDateTime. Cannot convert to TimestampNtz.",
                             prop.name));
         }
-        return TimestampNtz.fromLocalDateTime((LocalDateTime) v);
+        LocalDateTime ldt = (LocalDateTime) v;
+        LocalDateTime truncated = truncateToTimestampPrecision(ldt, precision);
+        return TimestampNtz.fromLocalDateTime(truncated);
     }
 
-    /** Converts an Instant or OffsetDateTime POJO property to Fluss TimestampLtz (UTC based). */
+    /**
+     * Converts an Instant or OffsetDateTime POJO property to Fluss TimestampLtz (UTC based),
+     * respecting the specified precision.
+     *
+     * @param precision the timestamp precision (0-9)
+     * @param prop the POJO property metadata
+     * @param v the value to convert
+     * @return TimestampLtz with precision applied, or null if v is null
+     */
     private static @Nullable TimestampLtz convertTimestampLtzValue(
-            PojoType.Property prop, @Nullable Object v) {
+            int precision, PojoType.Property prop, @Nullable Object v) {
         if (v == null) {
             return null;
         }
+        Instant instant;
         if (v instanceof Instant) {
-            return TimestampLtz.fromInstant((Instant) v);
+            instant = (Instant) v;
         } else if (v instanceof OffsetDateTime) {
-            return TimestampLtz.fromInstant(((OffsetDateTime) v).toInstant());
+            instant = ((OffsetDateTime) v).toInstant();
+        } else {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Field %s is not an Instant or OffsetDateTime. Cannot convert to TimestampLtz.",
+                            prop.name));
         }
-        throw new IllegalArgumentException(
-                String.format(
-                        "Field %s is not an Instant or OffsetDateTime. Cannot convert to TimestampLtz.",
-                        prop.name));
+        Instant truncated = truncateToTimestampPrecision(instant, precision);
+        return TimestampLtz.fromInstant(truncated);
+    }
+
+    /**
+     * Truncates a LocalDateTime to the specified timestamp precision.
+     *
+     * @param ldt the LocalDateTime to truncate
+     * @param precision the precision (0-9)
+     * @return truncated LocalDateTime
+     */
+    private static LocalDateTime truncateToTimestampPrecision(LocalDateTime ldt, int precision) {
+        if (precision >= 9) {
+            return ldt;
+        }
+        int nanos = ldt.getNano();
+        int truncatedNanos = truncateNanos(nanos, precision);
+        return ldt.withNano(truncatedNanos);
+    }
+
+    /**
+     * Truncates an Instant to the specified timestamp precision.
+     *
+     * @param instant the Instant to truncate
+     * @param precision the precision (0-9)
+     * @return truncated Instant
+     */
+    private static Instant truncateToTimestampPrecision(Instant instant, int precision) {
+        if (precision >= 9) {
+            return instant;
+        }
+        int nanos = instant.getNano();
+        int truncatedNanos = truncateNanos(nanos, precision);
+        return Instant.ofEpochSecond(instant.getEpochSecond(), truncatedNanos);
+    }
+
+    /**
+     * Truncates nanoseconds to the specified precision.
+     *
+     * @param nanos the nanoseconds value (0-999,999,999)
+     * @param precision the precision (0-9)
+     * @return truncated nanoseconds
+     */
+    private static int truncateNanos(int nanos, int precision) {
+        int divisor = (int) Math.pow(10, 9 - precision);
+        return (nanos / divisor) * divisor;
     }
 
     /** Converts a numeric POJO property to TINYINT (Byte). */