[pull] master from apache:master

.github/actions/setup-environment-action/action.yml

@@ -74,7 +74,7 @@ runs:
       uses: actions/setup-java@v3
       with:
         distribution: 'temurin'
-        java-version: ${{ inputs.java-version == 'default' && ((contains(github.job, 'Xlang') || contains(github.job, 'XVR') || contains(github.job, 'PreCommit_Java')) && '17' || '11') || inputs.java-version }}
+        java-version: ${{ inputs.java-version == 'default' && '11' || inputs.java-version }}
     - name: Setup Gradle
       uses: gradle/actions/setup-gradle@50e97c2cd7a37755bbfafc9c5b7cafaece252f6e # v6.1.0
       with:

.github/workflows/beam_PreCommit_Xlang_Generated_Transforms.yml

@@ -102,7 +102,7 @@ jobs:
       - name: Setup environment
         uses: ./.github/actions/setup-environment-action
         with:
-          java-version: '17'
+          java-version: default
           python-version: ${{ matrix.python_version }}
       - name: Set PY_VER_CLEAN
         id: set_py_ver_clean

CHANGES.md

@@ -67,6 +67,7 @@
 
 * Support for reading from Delta Lake added (Java) ([#38551](https://github.com/apache/beam/issues/38551)).
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
+* ClickHouseIO: support writing `DateTime64(precision[, 'timezone'])` columns with sub-second precision (Java) ([#38466](https://github.com/apache/beam/issues/38466)).
 
 ## New Features / Improvements
 

sdks/java/expansion-service/container/Dockerfile

@@ -16,7 +16,7 @@
 # limitations under the License.
 ###############################################################################
 
-FROM eclipse-temurin:17
+FROM eclipse-temurin:11
 LABEL Author "Apache Beam <dev@beam.apache.org>"
 ARG TARGETOS
 ARG TARGETARCH

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/transform/BeamBuiltinAggregations.java

@@ -83,6 +83,8 @@ public class BeamBuiltinAggregations {
                   typeName -> new DropNullFn(BeamBuiltinAggregations.createBitAnd(typeName)))
               .put("VAR_POP", t -> VarianceFn.newPopulation(t.getTypeName()))
               .put("VAR_SAMP", t -> VarianceFn.newSample(t.getTypeName()))
+              .put("STDDEV_POP", t -> VarianceFn.newPopulationStddev(t.getTypeName()))
+              .put("STDDEV_SAMP", t -> VarianceFn.newSampleStddev(t.getTypeName()))
               .put("COVAR_POP", t -> CovarianceFn.newPopulation(t.getTypeName()))
               .put("COVAR_SAMP", t -> CovarianceFn.newSample(t.getTypeName()))
               .put("COUNTIF", typeName -> CountIf.combineFn())

sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/transform/agg/VarianceFn.java

@@ -75,8 +75,12 @@ public class VarianceFn<T extends Number> extends Combine.CombineFn<T, VarianceA
   private static final boolean SAMPLE = true;
   private static final boolean POP = false;
 
-  private boolean isSample; // flag to determine return value should be Variance Pop or Sample
-  private SerializableFunction<BigDecimal, T> decimalConverter;
+  private final boolean isSample; // flag to determine return value should be Variance Pop or Sample
+  // When true, extractOutput returns the square root of the variance (i.e. standard deviation).
+  // Beam's enumerable bridge cannot translate a SQRT call layered on top of a window VAR_SAMP, so
+  // STDDEV_SAMP / STDDEV_POP are computed end-to-end inside this combiner instead.
+  private final boolean isStddev;
+  private final SerializableFunction<BigDecimal, T> decimalConverter;
 
   public static VarianceFn newPopulation(Schema.TypeName typeName) {
     return newPopulation(BigDecimalConverter.forSqlType(typeName));
@@ -85,7 +89,7 @@ public static VarianceFn newPopulation(Schema.TypeName typeName) {
   public static <V extends Number> VarianceFn newPopulation(
       SerializableFunction<BigDecimal, V> decimalConverter) {
 
-    return new VarianceFn<>(POP, decimalConverter);
+    return new VarianceFn<>(POP, false, decimalConverter);
   }
 
   public static VarianceFn newSample(Schema.TypeName typeName) {
@@ -95,11 +99,21 @@ public static VarianceFn newSample(Schema.TypeName typeName) {
   public static <V extends Number> VarianceFn newSample(
       SerializableFunction<BigDecimal, V> decimalConverter) {
 
-    return new VarianceFn<>(SAMPLE, decimalConverter);
+    return new VarianceFn<>(SAMPLE, false, decimalConverter);
   }
 
-  private VarianceFn(boolean isSample, SerializableFunction<BigDecimal, T> decimalConverter) {
+  public static VarianceFn newSampleStddev(Schema.TypeName typeName) {
+    return new VarianceFn<>(SAMPLE, true, BigDecimalConverter.forSqlType(typeName));
+  }
+
+  public static VarianceFn newPopulationStddev(Schema.TypeName typeName) {
+    return new VarianceFn<>(POP, true, BigDecimalConverter.forSqlType(typeName));
+  }
+
+  private VarianceFn(
+      boolean isSample, boolean isStddev, SerializableFunction<BigDecimal, T> decimalConverter) {
     this.isSample = isSample;
+    this.isStddev = isStddev;
     this.decimalConverter = decimalConverter;
   }
 
@@ -133,7 +147,19 @@ public Coder<VarianceAccumulator> getAccumulatorCoder(
 
   @Override
   public T extractOutput(VarianceAccumulator accumulator) {
-    return decimalConverter.apply(getVariance(accumulator));
+    BigDecimal result = getVariance(accumulator);
+    if (result != null && isStddev) {
+      double doubleVal = result.doubleValue();
+      if (doubleVal < 0.0) {
+        doubleVal = 0.0; // Clamp negative variance due to numerical instability
+      }
+      double sqrtVal = Math.sqrt(doubleVal);
+      if (Double.isInfinite(sqrtVal)) {
+        return decimalConverter.apply(result.sqrt(MATH_CTX));
+      }
+      result = BigDecimal.valueOf(sqrtVal);
+    }
+    return decimalConverter.apply(result);
   }
 
   private BigDecimal getVariance(VarianceAccumulator variance) {

sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/BeamSqlDslAggregationVarianceTest.java

@@ -30,7 +30,10 @@
 import org.junit.Rule;
 import org.junit.Test;
 
-/** Integration tests for {@code VAR_POP} and {@code VAR_SAMP}. */
+/**
+ * Integration tests for {@code VAR_POP}, {@code VAR_SAMP}, {@code STDDEV_POP} and {@code
+ * STDDEV_SAMP}.
+ */
 public class BeamSqlDslAggregationVarianceTest {
 
   private static final double PRECISION = 1e-7;
@@ -94,4 +97,42 @@ public void testSampleVarianceInt() {
 
     pipeline.run().waitUntilFinish();
   }
+
+  @Test
+  public void testPopulationStddevDouble() {
+    String sql = "SELECT STDDEV_POP(f_double) FROM PCOLLECTION GROUP BY f_int2";
+
+    PAssert.that(boundedInput.apply(SqlTransform.query(sql)))
+        .satisfies(matchesScalar(5.138887357, PRECISION));
+
+    pipeline.run().waitUntilFinish();
+  }
+
+  @Test
+  public void testPopulationStddevInt() {
+    String sql = "SELECT STDDEV_POP(f_int) FROM PCOLLECTION GROUP BY f_int2";
+
+    PAssert.that(boundedInput.apply(SqlTransform.query(sql))).satisfies(matchesScalar(5));
+
+    pipeline.run().waitUntilFinish();
+  }
+
+  @Test
+  public void testSampleStddevDouble() {
+    String sql = "SELECT STDDEV_SAMP(f_double) FROM PCOLLECTION GROUP BY f_int2";
+
+    PAssert.that(boundedInput.apply(SqlTransform.query(sql)))
+        .satisfies(matchesScalar(5.550632739, PRECISION));
+
+    pipeline.run().waitUntilFinish();
+  }
+
+  @Test
+  public void testSampleStddevInt() {
+    String sql = "SELECT STDDEV_SAMP(f_int) FROM PCOLLECTION GROUP BY f_int2";
+
+    PAssert.that(boundedInput.apply(SqlTransform.query(sql))).satisfies(matchesScalar(5));
+
+    pipeline.run().waitUntilFinish();
+  }
 }

sdks/java/extensions/sql/src/test/java/org/apache/beam/sdk/extensions/sql/impl/transform/agg/VarianceFnTest.java

@@ -26,6 +26,7 @@
 import java.util.Arrays;
 import org.apache.beam.sdk.coders.CoderRegistry;
 import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.schemas.Schema;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -51,18 +52,38 @@ public static Iterable<Object[]> varianceFns() {
             VarianceFn.newSample(BigDecimal::intValue),
             newVarianceAccumulator(FIFTEEN, FOUR, ZERO),
             5
+          },
+          {
+            VarianceFn.newPopulationStddev(Schema.TypeName.INT32),
+            newVarianceAccumulator(new BigDecimal(36), new BigDecimal(4), ZERO),
+            3
+          },
+          {
+            VarianceFn.newSampleStddev(Schema.TypeName.INT32),
+            newVarianceAccumulator(new BigDecimal(36), new BigDecimal(5), ZERO),
+            3
+          },
+          {
+            VarianceFn.newPopulationStddev(Schema.TypeName.DOUBLE),
+            newVarianceAccumulator(new BigDecimal("1e700"), BigDecimal.ONE, ZERO),
+            Double.POSITIVE_INFINITY
+          },
+          {
+            VarianceFn.newPopulationStddev(Schema.TypeName.FLOAT),
+            newVarianceAccumulator(new BigDecimal("1e700"), BigDecimal.ONE, ZERO),
+            Float.POSITIVE_INFINITY
           }
         });
   }
 
   private VarianceFn varianceFn;
   private VarianceAccumulator testAccumulatorInput;
-  private int expectedExtractedResult;
+  private Object expectedExtractedResult;
 
   public VarianceFnTest(
       VarianceFn varianceFn,
       VarianceAccumulator testAccumulatorInput,
-      int expectedExtractedResult) {
+      Object expectedExtractedResult) {
 
     this.varianceFn = varianceFn;
     this.testAccumulatorInput = testAccumulatorInput;

sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java

@@ -38,6 +38,8 @@
 import org.apache.beam.sdk.schemas.FieldAccessDescriptor;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes;
+import org.apache.beam.sdk.schemas.logicaltypes.NanosInstant;
+import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
 import org.apache.beam.sdk.schemas.transforms.Select;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -137,6 +139,7 @@
  * <tr><td>{@link TableSchema.TypeName#UINT64}</td> <td>{@link Schema.TypeName#INT64}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#DATE}</td> <td>{@link Schema.TypeName#DATETIME}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#DATETIME}</td> <td>{@link Schema.TypeName#DATETIME}</td></tr>
+ * <tr><td>{@link TableSchema.TypeName#DATETIME64}</td> <td>{@link Schema.TypeName#DATETIME} (precision &le; 3), {@link SqlTypes#TIMESTAMP} (4&ndash;6), or {@link NanosInstant} (&ge; 7)</td></tr>
  * <tr><td>{@link TableSchema.TypeName#ARRAY}</td> <td>{@link Schema.TypeName#ARRAY}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#ENUM8}</td> <td>{@link Schema.TypeName#STRING}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#ENUM16}</td> <td>{@link Schema.TypeName#STRING}</td></tr>

sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriter.java

@@ -39,6 +39,39 @@
 public class ClickHouseWriter {
   private static final Instant EPOCH_INSTANT = new Instant(0L);
 
+  // 10^0 through 10^9 inclusive — precision is validated in [0, 9] by ColumnType.dateTime64.
+  private static final long[] POW10 = {
+    1L, 10L, 100L, 1_000L, 10_000L, 100_000L, 1_000_000L, 10_000_000L, 100_000_000L, 1_000_000_000L
+  };
+
+  /**
+   * Encodes a timestamp into ClickHouse's {@code DateTime64(precision)} representation: a signed
+   * 64-bit integer counting ticks of size 10<sup>-precision</sup> seconds since the Unix epoch.
+   *
+   * <p>Accepts either a Joda {@link ReadableInstant} (millisecond precision) or a {@link
+   * java.time.Instant} (nanosecond precision). Sub-tick fractions are truncated toward negative
+   * infinity, matching ClickHouse's own encoding for negative timestamps.
+   */
+  static long encodeDateTime64(Object value, int precision) {
+    long epochSecond;
+    int nanoOfSecond;
+    if (value instanceof java.time.Instant) {
+      java.time.Instant inst = (java.time.Instant) value;
+      epochSecond = inst.getEpochSecond();
+      nanoOfSecond = inst.getNano();
+    } else if (value instanceof ReadableInstant) {
+      long millis = ((ReadableInstant) value).getMillis();
+      epochSecond = Math.floorDiv(millis, 1000L);
+      nanoOfSecond = (int) Math.floorMod(millis, 1000L) * 1_000_000;
+    } else {
+      throw new IllegalArgumentException(
+          "DateTime64 requires a Joda ReadableInstant or java.time.Instant, got "
+              + (value == null ? "null" : value.getClass().getName()));
+    }
+    long subSecondTicks = nanoOfSecond / POW10[9 - precision];
+    return Math.addExact(Math.multiplyExact(epochSecond, POW10[precision]), subSecondTicks);
+  }
+
   @SuppressWarnings("unchecked")
   static void writeNullableValue(ClickHouseOutputStream stream, ColumnType columnType, Object value)
       throws IOException {
@@ -138,6 +171,13 @@ static void writeValue(ClickHouseOutputStream stream, ColumnType columnType, Obj
         BinaryStreamUtils.writeUnsignedInt32(stream, epochSeconds);
         break;
 
+      case DATETIME64:
+        int precision =
+            Preconditions.checkNotNull(
+                columnType.precision(), "DateTime64 column is missing precision");
+        BinaryStreamUtils.writeInt64(stream, encodeDateTime64(value, precision));
+        break;
+
       case ARRAY:
         List<Object> values = (List<Object>) value;
         BinaryStreamUtils.writeVarInt(stream, values.size());