mxtymoshyk · Oct 25, 2026 · Oct 25, 2026
diff --git a/CHANGES.md b/CHANGES.md
@@ -68,6 +68,7 @@
 * (Java) FirestoreIO: added `RpcQosOptions.Builder.withRampupThrottlingDisabled()`, mirroring the existing DatastoreIO option, to opt out of the 500/50/5 write ramp-up ([#27170](https://github.com/apache/beam/issues/27170)).
 * Added support for attributing BigQuery API quota and billing to a specific GCP project (quota project): `quota_project_id` parameter in `ReadFromBigQuery` or `--quota_project_id` pipeline option, honored by the Python BigQuery IO transforms (`WriteToBigQuery` with `FILE_LOADS` or `STREAMING_INSERTS`; not yet `STORAGE_WRITE_API`, which is Java-backed), `--bigQueryQuotaProjectId` pipeline option (Java), and `bigqueryio.WithQuotaProject` read/query option (Go) ([#37431](https://github.com/apache/beam/issues/37431)).
 * (Java/Python/Go) DebeziumIO: `readAsJson()` can now include `op` and the remaining Debezium `source` fields (e.g. MySQL `file`, `pos`, `row`, `ts_ms`; PostgreSQL `lsn`, `txId`, `ts_ms`) in `metadata` via `withFullMetadata(true)` / `full_metadata=True` / `debeziumio.FullMetadata()`. Off by default ([#40701](https://github.com/apache/beam/issues/40701)).
+* (Java) BigQueryIO: added `Write.withRangePartitioning(...)` to create integer-range partitioned tables. For dynamic destinations, set a `RangePartitioning` in each `TableDestination` and call `withRangePartitioning()`. The option uses a new `TableDestinationCoderV4`, so to turn it on, drain the running pipeline and start a new one; in-place update won't work. `FILE_LOADS` temp tables now also copy the range partitioning of an existing destination table ([#27069](https://github.com/apache/beam/issues/27069)).
 
 ## New Features / Improvements
 

diff --git a/...ava/org/apache/beam/runners/dataflow/util/DefaultCoderCloudObjectTranslatorRegistrar.java b/...ava/org/apache/beam/runners/dataflow/util/DefaultCoderCloudObjectTranslatorRegistrar.java
@@ -45,6 +45,7 @@
 import org.apache.beam.sdk.io.gcp.bigquery.RowMutation;
 import org.apache.beam.sdk.io.gcp.bigquery.TableDestinationCoderV2;
 import org.apache.beam.sdk.io.gcp.bigquery.TableDestinationCoderV3;
+import org.apache.beam.sdk.io.gcp.bigquery.TableDestinationCoderV4;
 import org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
@@ -109,6 +110,7 @@ public class DefaultCoderCloudObjectTranslatorRegistrar
           StringUtf8Coder.class,
           TableDestinationCoderV2.class,
           TableDestinationCoderV3.class,
+          TableDestinationCoderV4.class,
           TableRowJsonCoder.class,
           TextualIntegerCoder.class,
           VarIntCoder.class,

diff --git a/...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java b/...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BatchLoads.java
@@ -188,6 +188,7 @@ class BatchLoads<DestinationT, ElementT>
       RowWriterFactory<ElementT, DestinationT> rowWriterFactory,
       @Nullable String kmsKey,
       boolean clusteringEnabled,
+      boolean rangePartitioningEnabled,
       boolean useAvroLogicalTypes,
       @Nullable String tempDataset,
       BadRecordRouter badRecordRouter,
@@ -217,7 +218,8 @@ class BatchLoads<DestinationT, ElementT>
     schemaUpdateOptions = Collections.emptySet();
     this.tempDataset = tempDataset;
     this.tableDestinationCoder =
-        clusteringEnabled ? TableDestinationCoderV3.of() : TableDestinationCoderV2.of();
+        DynamicDestinationsHelpers.tableDestinationCoder(
+            clusteringEnabled, rangePartitioningEnabled);
     this.badRecordRouter = badRecordRouter;
     this.badRecordErrorHandler = badRecordErrorHandler;
   }

diff --git a/...gle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java b/...gle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryHelpers.java
@@ -32,6 +32,7 @@
 import com.google.api.services.bigquery.model.JobConfigurationLoad;
 import com.google.api.services.bigquery.model.JobReference;
 import com.google.api.services.bigquery.model.JobStatus;
+import com.google.api.services.bigquery.model.RangePartitioning;
 import com.google.api.services.bigquery.model.Table;
 import com.google.api.services.bigquery.model.TableReference;
 import com.google.api.services.bigquery.model.TableSchema;
@@ -828,6 +829,13 @@ public String apply(TimePartitioning partitioning) {
     }
   }
 
+  static class RangePartitioningToJson implements SerializableFunction<RangePartitioning, String> {
+    @Override
+    public String apply(RangePartitioning partitioning) {
+      return toJsonString(partitioning);
+    }
+  }
+
   static Clustering clusteringFromJsonFields(String jsonStringClustering) {
     JsonElement jsonClustering = JsonParser.parseString(jsonStringClustering);
 

diff --git a/...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java b/...o/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java
@@ -33,6 +33,7 @@
 import com.google.api.services.bigquery.model.JobConfigurationQuery;
 import com.google.api.services.bigquery.model.JobReference;
 import com.google.api.services.bigquery.model.JobStatistics;
+import com.google.api.services.bigquery.model.RangePartitioning;
 import com.google.api.services.bigquery.model.Table;
 import com.google.api.services.bigquery.model.TableCell;
 import com.google.api.services.bigquery.model.TableConstraints;
@@ -94,6 +95,7 @@
 import org.apache.beam.sdk.io.fs.ResolveOptions;
 import org.apache.beam.sdk.io.fs.ResourceId;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.JsonTableRefToTableRef;
+import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.RangePartitioningToJson;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.TableRefToJson;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.TableSchemaToJsonSchema;
 import org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.TableSpecToTableRef;
@@ -431,6 +433,25 @@
  *         .withTimePartitioning(new TimePartitioning().setField("time")));
  * }</pre>
  *
+ * <h3>Writing to integer-range partitioned BigQuery tables</h3>
+ *
+ * <p>To create tables with <a
+ * href="https://cloud.google.com/bigquery/docs/partitioned-tables#integer_range">integer-range
+ * partitioning</a>, specify {@link BigQueryIO.Write#withRangePartitioning(RangePartitioning)}. For
+ * example:
+ *
+ * <pre>{@code
+ * PCollection<Order> orders = ...;
+ *
+ * orders.apply(BigQueryIO.write()
+ *         .withSchema(schema)
+ *         .withFormatFunction(order -> new TableRow()...)
+ *         .to("my-project:my_dataset.my_table")
+ *         .withRangePartitioning(new RangePartitioning()
+ *             .setField("customer_id")
+ *             .setRange(new RangePartitioning.Range().setStart(0L).setEnd(1000L).setInterval(10L))));
+ * }</pre>
+ *
  * <h3>Writing different values to different tables</h3>
  *
  * <p>A common use case is to dynamically generate BigQuery table names based on the current value.
@@ -2988,6 +3009,8 @@ public enum Method {
 
     abstract @Nullable ValueProvider<String> getJsonClustering();
 
+    abstract @Nullable ValueProvider<String> getJsonRangePartitioning();
+
     abstract CreateDisposition getCreateDisposition();
 
     abstract WriteDisposition getWriteDisposition();
@@ -3109,6 +3132,8 @@ abstract Builder<T> setAvroSchemaFactory(
 
       abstract Builder<T> setJsonClustering(ValueProvider<String> clustering);
 
+      abstract Builder<T> setJsonRangePartitioning(ValueProvider<String> jsonRangePartitioning);
+
       abstract Builder<T> setCreateDisposition(CreateDisposition createDisposition);
 
       abstract Builder<T> setWriteDisposition(WriteDisposition writeDisposition);
@@ -3330,7 +3355,8 @@ public Write<T> to(
      *
      * <p>If any of the returned destinations are configured with clustering fields, ensure that the
      * passed {@link DynamicDestinations} object returns {@link TableDestinationCoderV3} when {@link
-     * DynamicDestinations#getDestinationCoder()} is called.
+     * DynamicDestinations#getDestinationCoder()} is called. If any of the returned destinations are
+     * configured with range partitioning, return {@link TableDestinationCoderV4} instead.
      */
     public Write<T> to(DynamicDestinations<T, ?> dynamicDestinations) {
       checkArgument(dynamicDestinations != null, "dynamicDestinations can not be null");
@@ -3522,6 +3548,63 @@ public Write<T> withClustering() {
       return withClustering(new Clustering());
     }
 
+    /**
+     * Sets the <a
+     * href="https://cloud.google.com/bigquery/docs/partitioned-tables#integer_range">integer-range
+     * partitioning</a> of tables created by this write. BigQuery doesn't allow both time and range
+     * partitioning on one table, so this can't be combined with {@link #withTimePartitioning}.
+     *
+     * <p>This form is only for writing to a single table. With {@link #to(SerializableFunction)},
+     * call {@link #withRangePartitioning()} instead and set the range partitioning in each returned
+     * {@link TableDestination}. With {@link #to(DynamicDestinations)}, also return {@link
+     * TableDestinationCoderV4} from {@link DynamicDestinations#getDestinationCoder()}.
+     *
+     * <p>This option switches the destination coder to {@link TableDestinationCoderV4}, which also
+     * encodes range partitioning. V4 can't read state written by the older coders, so a running
+     * pipeline can't be updated in place to turn this on: drain it and start a new one. New
+     * pipelines can set it whether or not their tables are range partitioned.
+     */
+    public Write<T> withRangePartitioning(RangePartitioning partitioning) {
+      checkArgument(partitioning != null, "partitioning can not be null");
+      return withJsonRangePartitioning(
+          StaticValueProvider.of(BigQueryHelpers.toJsonString(partitioning)));
+    }
+
+    /**
+     * Like {@link #withRangePartitioning(RangePartitioning)} but using a deferred {@link
+     * ValueProvider}.
+     */
+    public Write<T> withRangePartitioning(ValueProvider<RangePartitioning> partitioning) {
+      checkArgument(partitioning != null, "partitioning can not be null");
+      return withJsonRangePartitioning(
+          NestedValueProvider.of(partitioning, new RangePartitioningToJson()));
+    }
+
+    /**
+     * The same as {@link #withRangePartitioning(RangePartitioning)}, but takes a JSON-serialized
+     * object. For example: {@code {"field": "id", "range": {"start": "0", "end": "100", "interval":
+     * "10"}}}
+     */
+    public Write<T> withJsonRangePartitioning(ValueProvider<String> partitioning) {
+      checkArgument(partitioning != null, "partitioning can not be null");
+      return toBuilder().setJsonRangePartitioning(partitioning).build();
+    }
+
+    /**
+     * Turns on range partitioning for dynamic destinations. Each {@link TableDestination} returned
+     * by {@link #to(SerializableFunction)} sets its own range partitioning. With {@link
+     * #to(DynamicDestinations)}, also return {@link TableDestinationCoderV4} from {@link
+     * DynamicDestinations#getDestinationCoder()}. For a single table, use {@link
+     * #withRangePartitioning(RangePartitioning)}.
+     *
+     * <p>Like {@link #withRangePartitioning(RangePartitioning)}, this switches to {@link
+     * TableDestinationCoderV4}, so a running pipeline has to be drained and restarted to turn it
+     * on.
+     */
+    public Write<T> withRangePartitioning() {
+      return withRangePartitioning(new RangePartitioning());
+    }
+
     /** Specifies whether the table should be created if it does not exist. */
     public Write<T> withCreateDisposition(CreateDisposition createDisposition) {
       checkArgument(createDisposition != null, "createDisposition can not be null");
@@ -4219,6 +4302,29 @@ && getStorageApiTriggeringFrequency(bqOptions) != null) {
             !getUseBeamSchema(), "Auto schema update not supported when using Beam schemas.");
       }
 
+      checkArgument(
+          !(hasJsonConfig(getJsonTimePartitioning()) && hasJsonConfig(getJsonRangePartitioning())),
+          "A BigQuery table can be either time partitioned or range partitioned. Only one of"
+              + " withTimePartitioning and withRangePartitioning may be set.");
+
+      ValueProvider<String> jsonRangePartitioning = getJsonRangePartitioning();
+      if (jsonRangePartitioning != null
+          && (getDynamicDestinations() != null || getTableFunction() != null)) {
+        checkArgument(
+            jsonRangePartitioning.isAccessible()
+                && !ConstantTimePartitioningClusteringDestinations.isJsonConfigPresent(
+                    jsonRangePartitioning),
+            "When writing to dynamic destinations, set range partitioning in the returned"
+                + " TableDestination objects and call withRangePartitioning() with no arguments.");
+      } else if (jsonRangePartitioning != null && jsonRangePartitioning.isAccessible()) {
+        checkArgument(
+            ConstantTimePartitioningClusteringDestinations.isJsonConfigPresent(
+                jsonRangePartitioning),
+            "withRangePartitioning() with no arguments only applies to to(SerializableFunction)"
+                + " and to(DynamicDestinations). To write to a single table, pass a"
+                + " RangePartitioning.");
+      }
+
       if (getJsonTimePartitioning() != null) {
         checkArgument(
             getDynamicDestinations() == null,
@@ -4240,13 +4346,18 @@ && getStorageApiTriggeringFrequency(bqOptions) != null) {
         if (jsonTableRef != null) {
           tableDestinations =
               DynamicDestinationsHelpers.ConstantTableDestinations.fromJsonTableRef(
-                  jsonTableRef, getTableDescription(), jsonClustering != null);
+                  jsonTableRef,
+                  getTableDescription(),
+                  jsonClustering != null,
+                  jsonRangePartitioning != null);
         } else {
           // Checked above: exactly one of jsonTableRef, tableFunction or dynamicDestinations is
           // set.
           tableDestinations =
               new TableFunctionDestinations<>(
-                  checkStateNotNull(tableFunction), jsonClustering != null);
+                  checkStateNotNull(tableFunction),
+                  jsonClustering != null,
+                  jsonRangePartitioning != null);
         }
 
         // Wrap with a DynamicDestinations class that will provide a schema. There might be no
@@ -4259,12 +4370,15 @@ && getStorageApiTriggeringFrequency(bqOptions) != null) {
           tableDestinations = new SchemaFromViewDestinations<>(tableDestinations, schemaFromView);
         }
 
-        // Wrap with a DynamicDestinations class that will provide the proper TimePartitioning.
+        // Wrap with a DynamicDestinations class that will provide the proper TimePartitioning,
+        // Clustering and RangePartitioning.
         ValueProvider<String> jsonTimePartitioning = getJsonTimePartitioning();
-        if (jsonTimePartitioning != null || jsonClustering != null) {
+        if (jsonTimePartitioning != null
+            || jsonClustering != null
+            || jsonRangePartitioning != null) {
           tableDestinations =
               new ConstantTimePartitioningClusteringDestinations<>(
-                  tableDestinations, jsonTimePartitioning, jsonClustering);
+                  tableDestinations, jsonTimePartitioning, jsonClustering, jsonRangePartitioning);
         }
         List<String> primaryKey = getPrimaryKey();
         if (primaryKey != null) {
@@ -4403,7 +4517,10 @@ private <DestinationT> WriteResult expandTyped(
           input
               .apply(
                   "PrepareWrite",
-                  new PrepareWrite<>(dynamicDestinations, SerializableFunctions.identity()))
+                  new PrepareWrite<>(
+                      dynamicDestinations,
+                      SerializableFunctions.identity(),
+                      PrepareWrite.dropsRangePartitioning(destinationCoder)))
               .setCoder(KvCoder.of(destinationCoder, input.getCoder()));
 
       return continueExpandTyped(
@@ -4530,6 +4647,7 @@ private <DestinationT> WriteResult continueExpandTyped(
                 rowWriterFactory,
                 getKmsKey(),
                 getJsonClustering() != null,
+                getJsonRangePartitioning() != null,
                 getUseAvroLogicalTypes(),
                 getWriteTempDataset(),
                 getBadRecordRouter(),
@@ -4715,6 +4833,13 @@ private boolean hasJsonTypeInSchema(JsonElement schema) {
       return false;
     }
 
+    /** Returns true if the JSON config is accessible at construction time and not empty. */
+    private static boolean hasJsonConfig(@Nullable ValueProvider<String> json) {
+      return json != null
+          && json.isAccessible()
+          && ConstantTimePartitioningClusteringDestinations.isJsonConfigPresent(json);
+    }
+
     @Override
     public void populateDisplayData(DisplayData.Builder builder) {
       super.populateDisplayData(builder);

diff --git a/...oud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java b/...oud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOTranslation.java
@@ -483,6 +483,7 @@ static class BigQueryIOWriteTranslator implements TransformPayloadTranslator<Wri
             .addNullableByteArrayField("bad_record_error_handler")
             .addNullableByteArrayField("bad_record_router")
             .addNullableInt64Field("auto_schema_update_strict_timeout_ms")
+            .addNullableStringField("json_range_partitioning")
             .build();
 
     public static final String BIGQUERY_WRITE_TRANSFORM_URN =
@@ -557,6 +558,10 @@ public Row toConfigRow(Write<?> transform) {
       if (jsonClustering != null) {
         fieldValues.put("clustering", jsonClustering.get());
       }
+      ValueProvider<String> jsonRangePartitioning = transform.getJsonRangePartitioning();
+      if (jsonRangePartitioning != null) {
+        fieldValues.put("json_range_partitioning", jsonRangePartitioning.get());
+      }
       if (transform.getCreateDisposition() != null) {
         fieldValues.put("create_disposition", toByteArray(transform.getCreateDisposition()));
       }
@@ -749,6 +754,15 @@ public Write<?> fromConfigRow(Row configRow, PipelineOptions options) {
             builder = builder.setJsonClustering(StaticValueProvider.of(jsonClustering));
           }
         }
+        if (TransformUpgrader.compareVersions(updateCompatibilityBeamVersion, "2.78.0") >= 0) {
+          // This property was added for Beam 2.78.0 hence not available when
+          // upgrading the transform from previous Beam versions.
+          String jsonRangePartitioning = configRow.getString("json_range_partitioning");
+          if (jsonRangePartitioning != null) {
+            builder =
+                builder.setJsonRangePartitioning(StaticValueProvider.of(jsonRangePartitioning));
+          }
+        }
         byte[] createDispositionBytes = configRow.getBytes("create_disposition");
         if (createDispositionBytes != null) {
           builder =

diff --git a/...-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/CreateTableHelpers.java b/...-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/CreateTableHelpers.java
@@ -27,6 +27,7 @@
 import com.google.api.services.bigquery.model.BigLakeConfiguration;
 import com.google.api.services.bigquery.model.Clustering;
 import com.google.api.services.bigquery.model.EncryptionConfiguration;
+import com.google.api.services.bigquery.model.RangePartitioning;
 import com.google.api.services.bigquery.model.Table;
 import com.google.api.services.bigquery.model.TableConstraints;
 import com.google.api.services.bigquery.model.TableReference;
@@ -204,6 +205,11 @@ private static void tryCreateTable(
           table.setClustering(clustering);
         }
 
+        RangePartitioning rangePartitioning = tableDestination.getRangePartitioning();
+        if (rangePartitioning != null) {
+          table.setRangePartitioning(rangePartitioning);
+        }
+
         if (kmsKey != null) {
           table.setEncryptionConfiguration(new EncryptionConfiguration().setKmsKeyName(kmsKey));
         }