You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/08/05 07:49:10 UTC
[L] Change in asterixdb[master]: MB-72415: Invalidate query plan cache on REST link DDL
From Rithwik Koul <ri...@couchbase.com>:
Rithwik Koul has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21493?usp=email )
Change subject: MB-72415: Invalidate query plan cache on REST link DDL
......................................................................
MB-72415: Invalidate query plan cache on REST link DDL
Clear the cache when a link DDL request commits a metadata change,
covering POST, PUT and DELETE through their common entry point. Requests
that fail leave the cache untouched.
Change-Id: Ib409d6c9022f36a94946796aa22d772028a35659
---
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
M asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
A asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
A asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java
M hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupWriteOperatorNodePushable.java
A hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/RetainingFrameOptimizedHybridHashJoinTest.java
31 files changed, 818 insertions(+), 16 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/93/21493/1
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp
new file mode 100644
index 0000000..68b9812
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp
@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test if exists;
+CREATE DATAVERSE test;
+USE test;
+
+CREATE TYPE ColumnType1 AS {
+ id: integer
+ };
+
+CREATE COLLECTION snapshots(ColumnType1) PRIMARY KEY id;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp
new file mode 100644
index 0000000..c10705d
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+/*
+ * Each object is roughly 3KB, so 4000 objects write well past the 5MB limit used in the COPY TO below
+ * and the output has to be split over several files.
+ */
+
+USE test;
+
+INSERT INTO snapshots(
+ SELECT VALUE {
+ "id": intVal,
+ "padding_a": repeat("a", 1000),
+ "padding_b": repeat("b", 1000),
+ "padding_c": repeat("c", 1000)
+ }
+ FROM range(1, 4000) intVal
+);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp
new file mode 100644
index 0000000..84d20da
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+ SELECT c.* FROM snapshots c ORDER BY c.id
+ ) toWriter
+TO %adapter%
+PATH (%pathprefix% "copy-to-result", "max-file-size")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "5MB"
+ };
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp
new file mode 100644
index 0000000..17f7622
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE TYPE ColumnType2 AS {
+ };
+
+CREATE EXTERNAL DATASET TestDataset(ColumnType2) USING %adapter%
+(
+ %template%,
+ %additional_Properties%
+ ("definition"="%path_prefix%copy-to-result/max-file-size/"),
+ ("include"="*.json"),
+ ("requireVersionChangeDetection"="false"),
+ ("format" = "json")
+);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp
new file mode 100644
index 0000000..a6965e6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+/*
+ * Splitting the output over several files must not lose or duplicate any object, so reading all of them
+ * back has to reproduce the source collection exactly.
+ */
+
+USE test;
+
+SELECT VALUE {
+ "objects": count(c.id),
+ "distinct_ids": count(DISTINCT c.id),
+ "id_sum": sum(c.id)
+}
+FROM TestDataset c;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp
new file mode 100644
index 0000000..6f4cc98
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+CREATE TYPE OpenType AS {
+};
+
+CREATE EXTERNAL DATASET Customer(OpenType) USING %adapter% (
+ %template%,
+ %additional_Properties%
+ ("definition"="%path_prefix%external-filter/car/{company:string}/customer/{customer_id:int}"),
+ ("embed-filter-values" = "false"),
+ ("format"="json")
+);
+
+
+
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp
new file mode 100644
index 0000000..7146a1b
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY Customer c
+TO %adapter%
+PATH (%pathprefix% "copy-to-result-with-over")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "1MB"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp
new file mode 100644
index 0000000..101453f
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY Customer c
+TO %adapter%
+PATH (%pathprefix% "copy-to-result-with-over")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "random"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp
new file mode 100644
index 0000000..9f11603
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY Customer c
+TO %adapter%
+PATH (%pathprefix% "copy-to-result-with-over")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "6TB"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm
new file mode 100644
index 0000000..2992765
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm
@@ -0,0 +1 @@
+{ "objects": 4000, "distinct_ids": 4000, "id_sum": 8002000 }
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
index dc56157..0efb85a 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
@@ -274,6 +274,19 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to/negative">
+ <compilation-unit name="bad-max-file-size">
+ <placeholder name="adapter" value="AZUREBLOB" />
+ <placeholder name="pathprefix" value="" />
+ <placeholder name="path_prefix" value="" />
+ <placeholder name="additionalProperties" value='"container":"playground",' />
+ <placeholder name="additional_Properties" value='("container"="playground"),' />
+ <output-dir compare="Text">bad-max-file-size</output-dir>
+ <expected-error>Minimum value allowed for 'max-file-size' is 5 MiB. Found 1MB</expected-error>
+ <expected-error>ASX1206: Storage units expected for the field 'max-file-size' (e.g., 0.1KB, 100kb, 1mb, 3MB, 8.5GB ...). Provided 'random'</expected-error>
+ <expected-error>Maximum value allowed for 'max-file-size' is 5 TiB. Found 6TB</expected-error>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to/negative">
<compilation-unit name="csv-error-checks">
<placeholder name="adapter" value="AZUREBLOB" />
<placeholder name="pathprefix" value="" />
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
index 10b11cc..c4c395d 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
@@ -190,6 +190,16 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to">
+ <compilation-unit name="max-file-size">
+ <placeholder name="adapter" value="S3" />
+ <placeholder name="pathprefix" value="" />
+ <placeholder name="path_prefix" value="" />
+ <placeholder name="additionalProperties" value='"container":"playground",' />
+ <placeholder name="additional_Properties" value='("container"="playground"),' />
+ <output-dir compare="Text">max-file-size</output-dir>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to">
<compilation-unit name="parquet-field-names">
<placeholder name="adapter" value="S3" />
<placeholder name="pathprefix" value="" />
@@ -373,6 +383,19 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to/negative">
+ <compilation-unit name="bad-max-file-size">
+ <placeholder name="adapter" value="S3" />
+ <placeholder name="pathprefix" value="" />
+ <placeholder name="path_prefix" value="" />
+ <placeholder name="additionalProperties" value='"container":"playground",' />
+ <placeholder name="additional_Properties" value='("container"="playground"),' />
+ <output-dir compare="Text">bad-max-file-size</output-dir>
+ <expected-error>Minimum value allowed for 'max-file-size' is 5 MiB. Found 1MB</expected-error>
+ <expected-error>ASX1206: Storage units expected for the field 'max-file-size' (e.g., 0.1KB, 100kb, 1mb, 3MB, 8.5GB ...). Provided 'random'</expected-error>
+ <expected-error>Maximum value allowed for 'max-file-size' is 5 TiB. Found 6TB</expected-error>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to/negative">
<compilation-unit name="csv-error-checks">
<placeholder name="adapter" value="S3" />
<placeholder name="pathprefix" value="" />
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
index c0635af..f50169c 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
@@ -293,6 +293,20 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to/negative">
+ <compilation-unit name="bad-max-file-size">
+ <placeholder name="adapter" value="HDFS" />
+ <placeholder name="pathprefix" value='"/playground", ' />
+ <placeholder name="path_prefix" value="/playground/" />
+ <placeholder name="additionalProperties" value="" />
+ <placeholder name="additional_Properties" value='("input-format" = "text-input-format"),' />
+ <output-dir compare="Text">bad-max-file-size</output-dir>
+ <expected-error>Minimum value allowed for 'max-file-size' is 5 MiB. Found 1MB</expected-error>
+ <expected-error>ASX1206: Storage units expected for the field 'max-file-size' (e.g., 0.1KB, 100kb, 1mb, 3MB, 8.5GB ...). Provided 'random'</expected-error>
+ <expected-error>Maximum value allowed for 'max-file-size' is 5 TiB. Found 6TB</expected-error>
+ <source-location>false</source-location>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to/negative">
<compilation-unit name="csv-error-checks">
<placeholder name="adapter" value="HDFS" />
<placeholder name="pathprefix" value='"/playground", ' />
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
index 5d9ab7f..f3de8a8 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
@@ -38,15 +38,17 @@
private static final long serialVersionUID = 8971234908711236L;
private final IExternalFileWriterFactory writerFactory;
private final int maxResult;
+ private final long maxFileSize;
private final ParquetExternalFilePrinterFactory printerFactory;
private final IPathResolver resolver;
private final IHyracksTaskContext ctx;
public ParquetExternalWriterFactory(IHyracksTaskContext ctx, IExternalFileWriterFactory writerFactory,
- int maxResult, ParquetExternalFilePrinterFactory printerFactory, IPathResolver resolver) {
+ int maxResult, long maxFileSize, ParquetExternalFilePrinterFactory printerFactory, IPathResolver resolver) {
this.ctx = ctx;
this.writerFactory = writerFactory;
this.maxResult = maxResult;
+ this.maxFileSize = maxFileSize;
this.printerFactory = printerFactory;
this.resolver = resolver;
}
@@ -55,7 +57,7 @@
MessageType schema = generateSchema(schemaNode);
printerFactory.setParquetSchema(schema);
IExternalFileWriter writer = writerFactory.createWriter(ctx, printerFactory);
- return new ExternalFileWriter(resolver, writer, maxResult);
+ return new ExternalFileWriter(resolver, writer, maxResult, maxFileSize);
}
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
index 3a276a2..04dc860 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
@@ -39,12 +39,13 @@
private final IAType sourceType;
private final IExternalFileWriterFactory writerFactory;
private final int maxResult;
+ private final long maxFileSize;
private final ParquetExternalFilePrinterFactoryProvider printerFactoryProvider;
private final IPathResolverFactory pathResolverFactory;
public ParquetSinkExternalWriterFactory(WriterPartitionerFactory partitionerFactory,
RecordDescriptor inputRecordDesc, int sourceColumn, IAType sourceType, int maxSchemas,
- IExternalFileWriterFactory writerFactory, int maxResult,
+ IExternalFileWriterFactory writerFactory, int maxResult, long maxFileSize,
ParquetExternalFilePrinterFactoryProvider printerFactoryProvider,
IPathResolverFactory pathResolverFactory) {
this.partitionerFactory = partitionerFactory;
@@ -54,6 +55,7 @@
this.maxSchemas = maxSchemas;
this.writerFactory = writerFactory;
this.maxResult = maxResult;
+ this.maxFileSize = maxFileSize;
this.printerFactoryProvider = printerFactoryProvider;
this.pathResolverFactory = pathResolverFactory;
}
@@ -63,7 +65,7 @@
ParquetExternalFilePrinterFactory printerFactory =
(ParquetExternalFilePrinterFactory) printerFactoryProvider.createPrinterFactory();
ParquetExternalWriterFactory parquetExternalWriterFactory = new ParquetExternalWriterFactory(ctx, writerFactory,
- maxResult, printerFactory, pathResolverFactory.createResolver(ctx));
+ maxResult, maxFileSize, printerFactory, pathResolverFactory.createResolver(ctx));
ParquetSinkExternalWriterRuntime runtime =
new ParquetSinkExternalWriterRuntime(sourceColumn, partitionerFactory.createPartitioner(),
inputRecordDesc, parquetExternalWriterFactory, sourceType, maxSchemas);
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
index e389c03..e328b4f 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
@@ -107,6 +107,12 @@
}
@Override
+ public final long getBytesWritten() {
+ long acceptedBytes = cloudWriter == null ? 0 : cloudWriter.position();
+ return Math.max(acceptedBytes, printer.getBytesWritten());
+ }
+
+ @Override
public final void abort() throws HyracksDataException {
try {
if (cloudWriter != null) {
diff --git a/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md b/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
index 4f7174c..c149d5a 100644
--- a/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
+++ b/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
@@ -777,6 +777,7 @@
"format" : "json",
"compression": "gzip",
"max-objects-per-file": 1000,
+ "max-file-size": "128MB",
"container": "myBucket",
"accessKeyId": "<access-key>",
"secretAccessKey": "<secret-key>",
@@ -784,5 +785,13 @@
};
This query will be exported as partitions into a set of folders, with one folder for each value of `battery_status`.
-Each partition itself will also be sorted by the `name` field, and compressed with `gzip` and divided into files of 100
+Each partition itself will also be sorted by the `name` field, and compressed with `gzip` and divided into files of 1000
objects or fewer per file.
+
+A file is also started afresh once it reaches `max-file-size`, so whichever of the two limits is reached first ends the
+current file. `max-file-size` accepts a storage unit string such as `5MB` or `1GB`, has a minimum of `5MB`, and is
+unbounded when it is not specified. It is measured on the bytes written to the target after compression and is
+approximate: an object is never split across files, so the object that reaches the limit takes the file past it. For
+`parquet` the size is only observable at row group boundaries, so `max-file-size` should be set comfortably above
+`row-group-size`. The largest file the target itself accepts is generally lower than the maximum accepted here, as it
+depends on the target's multipart upload limits.
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index e12991f..d115e2f 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -379,10 +379,16 @@
* Writer Constants
*/
public static final String KEY_WRITER_MAX_RESULT = "max-objects-per-file";
+ public static final String KEY_WRITER_MAX_FILE_SIZE = "max-file-size";
public static final String KEY_VALIDATE_WRITE_PERMISSION = "validate-write-permission";
public static final String KEY_WRITER_COMPRESSION = "compression";
public static final int WRITER_MAX_RESULT_DEFAULT = 10000;
public static final int WRITER_MAX_RESULT_MINIMUM = 1000;
+ public static final long WRITER_MAX_FILE_SIZE_UNLIMITED = -1L;
+ public static final long WRITER_MAX_FILE_SIZE_MINIMUM =
+ StorageUtil.getLongSizeInBytes(5, StorageUtil.StorageUnit.MEGABYTE);
+ public static final long WRITER_MAX_FILE_SIZE_MAXIMUM =
+ StorageUtil.getLongSizeInBytes(5, StorageUtil.StorageUnit.TERABYTE);
public static final Set<String> WRITER_SUPPORTED_FORMATS;
public static final Set<String> WRITER_SUPPORTED_ADAPTERS;
public static final Set<String> TEXTUAL_WRITER_SUPPORTED_COMPRESSION;
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
index 92afd40..0bb2bee 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
@@ -31,10 +31,13 @@
import static org.apache.asterix.external.util.ExternalDataConstants.FORMAT_PARQUET;
import static org.apache.asterix.external.util.ExternalDataConstants.KEY_PARQUET_PAGE_SIZE;
import static org.apache.asterix.external.util.ExternalDataConstants.KEY_PARQUET_ROW_GROUP_SIZE;
+import static org.apache.asterix.external.util.ExternalDataConstants.KEY_WRITER_MAX_FILE_SIZE;
import static org.apache.asterix.external.util.ExternalDataConstants.KEY_WRITER_MAX_RESULT;
import static org.apache.asterix.external.util.ExternalDataConstants.PARQUET_MAX_SCHEMAS_KEY;
import static org.apache.asterix.external.util.ExternalDataConstants.PARQUET_MAX_SCHEMAS_MAX_VALUE;
import static org.apache.asterix.external.util.ExternalDataConstants.PARQUET_WRITER_VERSION_KEY;
+import static org.apache.asterix.external.util.ExternalDataConstants.WRITER_MAX_FILE_SIZE_MAXIMUM;
+import static org.apache.asterix.external.util.ExternalDataConstants.WRITER_MAX_FILE_SIZE_MINIMUM;
import static org.apache.asterix.external.util.ExternalDataConstants.WRITER_MAX_RESULT_MINIMUM;
import java.util.List;
@@ -57,6 +60,7 @@
validateAdapter(adapter, supportedAdapters, sourceLocation);
validateFormat(configuration, sourceLocation);
validateMaxResult(configuration, sourceLocation);
+ validateMaxFileSize(configuration);
}
private static void validateQuote(Map<String, String> configuration, SourceLocation sourceLocation)
@@ -183,6 +187,29 @@
}
}
+ private static void validateMaxFileSize(Map<String, String> configuration) throws CompilationException {
+ String maxFileSize = configuration.get(KEY_WRITER_MAX_FILE_SIZE);
+ if (maxFileSize == null) {
+ return;
+ }
+
+ long value;
+ try {
+ value = StorageUtil.getByteValue(maxFileSize);
+ } catch (IllegalArgumentException e) {
+ throw CompilationException.create(ErrorCode.ILLEGAL_SIZE_PROVIDED, KEY_WRITER_MAX_FILE_SIZE, maxFileSize);
+ }
+
+ if (value < WRITER_MAX_FILE_SIZE_MINIMUM) {
+ throw new CompilationException(MINIMUM_VALUE_ALLOWED_FOR_PARAM, KEY_WRITER_MAX_FILE_SIZE,
+ StorageUtil.toHumanReadableSize(WRITER_MAX_FILE_SIZE_MINIMUM), maxFileSize);
+ }
+ if (value > WRITER_MAX_FILE_SIZE_MAXIMUM) {
+ throw new CompilationException(MAXIMUM_VALUE_ALLOWED_FOR_PARAM, KEY_WRITER_MAX_FILE_SIZE,
+ StorageUtil.toHumanReadableSize(WRITER_MAX_FILE_SIZE_MAXIMUM), maxFileSize);
+ }
+ }
+
private static void validateMaxParquetSchemas(Map<String, String> configuration, SourceLocation sourceLocation)
throws CompilationException {
String maxResult = configuration.get(PARQUET_MAX_SCHEMAS_KEY);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
index 510c8e4..e8f7e63 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
@@ -109,6 +109,12 @@
}
@Override
+ public long getBytesWritten() {
+ long writtenBytes = outputStream == null ? 0 : outputStream.getPos();
+ return Math.max(writtenBytes, printer.getBytesWritten());
+ }
+
+ @Override
public void abort() throws HyracksDataException {
try {
printer.close();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
index 906e272..e1d0bd4 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
@@ -36,6 +36,7 @@
private final IExternalPrinter printer;
private final ILocalFSValidator validator;
private final SourceLocation pathSourceLocation;
+ private FileOutputStream currentFileStream;
LocalFSExternalFileWriter(IExternalPrinter printer, ILocalFSValidator validator,
SourceLocation pathSourceLocation) {
@@ -64,7 +65,8 @@
if (!currentFile.createNewFile()) {
throw RuntimeDataException.create(ErrorCode.COULD_NOT_CREATE_FILE, currentFile.getAbsolutePath());
}
- printer.newStream(new BufferedOutputStream(new FileOutputStream(currentFile)));
+ currentFileStream = new FileOutputStream(currentFile);
+ printer.newStream(new BufferedOutputStream(currentFileStream));
} catch (IOException e) {
throw HyracksDataException.create(e);
}
@@ -77,6 +79,17 @@
}
@Override
+ public long getBytesWritten() throws HyracksDataException {
+ try {
+ // lags behind by whatever the buffered stream has not flushed yet
+ long writtenBytes = currentFileStream == null ? 0 : currentFileStream.getChannel().position();
+ return Math.max(writtenBytes, printer.getBytesWritten());
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ @Override
public void abort() throws HyracksDataException {
printer.close();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
index 3ef41b9..becfe30 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
@@ -82,6 +82,13 @@
}
@Override
+ public long getBytesWritten() {
+ // flushed row groups are counted as compressed bytes, the open row group as encoded bytes; the
+ // footer is only written on close and is therefore not counted
+ return writer == null ? 0 : writer.getDataSize();
+ }
+
+ @Override
public void print(IValueReference value) throws HyracksDataException {
try {
this.writer.write(value);
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java
new file mode 100644
index 0000000..13966af
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.
+ */
+package org.apache.asterix.external.util;
+
+import static org.apache.asterix.external.util.ExternalDataConstants.KEY_WRITER_MAX_FILE_SIZE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.asterix.common.exceptions.CompilationException;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.junit.Test;
+
+public class WriterMaxFileSizeValidationTest {
+
+ private static final String ADAPTER = ExternalDataConstants.KEY_ADAPTER_NAME_AWS_S3;
+
+ @Test
+ public void testAbsentIsAccepted() throws CompilationException {
+ validate(null);
+ }
+
+ @Test
+ public void testValidSizesAreAccepted() throws CompilationException {
+ validate("5MB");
+ validate("128MB");
+ validate("1GB");
+ validate("8.5GB");
+ validate("5TB");
+ validate("10485760");
+ }
+
+ @Test
+ public void testUnparsableSizeIsRejected() {
+ assertRejected("random", ErrorCode.ILLEGAL_SIZE_PROVIDED);
+ assertRejected("", ErrorCode.ILLEGAL_SIZE_PROVIDED);
+ assertRejected("10XB", ErrorCode.ILLEGAL_SIZE_PROVIDED);
+ }
+
+ @Test
+ public void testSizeBelowMinimumIsRejected() {
+ assertRejected("1MB", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("1024", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("0", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ }
+
+ @Test
+ public void testNegativeSizeIsRejected() {
+ // the size parser accepts a leading minus, so the minimum check is what rules these out
+ assertRejected("-1", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("-10MB", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ }
+
+ @Test
+ public void testSizeAboveMaximumIsRejected() {
+ assertRejected("6TB", ErrorCode.MAXIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("1PB", ErrorCode.MAXIMUM_VALUE_ALLOWED_FOR_PARAM);
+ }
+
+ @Test
+ public void testRejectionNamesTheOption() {
+ try {
+ validate("random");
+ fail("expected the invalid size to be rejected");
+ } catch (CompilationException e) {
+ assertTrue("error message does not name the option: " + e.getMessage(),
+ e.getMessage().contains(KEY_WRITER_MAX_FILE_SIZE));
+ }
+ }
+
+ private static void assertRejected(String maxFileSize, ErrorCode expected) {
+ try {
+ validate(maxFileSize);
+ fail("expected '" + maxFileSize + "' to be rejected");
+ } catch (CompilationException e) {
+ assertEquals("unexpected error for '" + maxFileSize + "'", expected.intValue(), e.getErrorCode());
+ }
+ }
+
+ private static void validate(String maxFileSize) throws CompilationException {
+ Map<String, String> configuration = new HashMap<>();
+ configuration.put(ExternalDataConstants.KEY_FORMAT, ExternalDataConstants.FORMAT_JSON_LOWER_CASE);
+ if (maxFileSize != null) {
+ configuration.put(KEY_WRITER_MAX_FILE_SIZE, maxFileSize);
+ }
+ WriterValidationUtil.validateWriterConfiguration(ADAPTER, ExternalDataConstants.WRITER_SUPPORTED_ADAPTERS,
+ configuration, null);
+ }
+}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
index 8121f6a..5ba4325 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
@@ -116,6 +116,14 @@
return Integer.parseInt(maxResultString);
}
+ private static long getMaxFileSize(IWriteDataSink sink) {
+ String maxFileSizeString = sink.getConfiguration().get(ExternalDataConstants.KEY_WRITER_MAX_FILE_SIZE);
+ if (maxFileSizeString == null) {
+ return ExternalDataConstants.WRITER_MAX_FILE_SIZE_UNLIMITED;
+ }
+ return StorageUtil.getByteValue(maxFileSizeString);
+ }
+
private static int getMaxParquetSchema(Map<String, String> conf) {
String maxResultString = conf.get(ExternalDataConstants.PARQUET_MAX_SCHEMAS_KEY);
if (maxResultString == null) {
@@ -159,6 +167,7 @@
fileWriterFactory.validate(appCtx);
String fileExtension = ExternalWriterProvider.getFileExtension(sink);
int maxResult = ExternalWriterProvider.getMaxResult(sink);
+ long maxFileSize = ExternalWriterProvider.getMaxFileSize(sink);
Map<String, String> configuration = sink.getConfiguration();
String format = configuration.get(ExternalDataConstants.KEY_FORMAT);
@@ -184,7 +193,7 @@
printerFactory = CleanJSONPrinterFactoryProvider.INSTANCE.getPrinterFactory(sourceType);
externalPrinterFactory = new TextualExternalFilePrinterFactory(printerFactory, compressStreamFactory);
writerFactory = new ExternalFileWriterFactory(fileWriterFactory, externalPrinterFactory,
- pathResolverFactory, maxResult);
+ pathResolverFactory, maxResult, maxFileSize);
return new SinkExternalWriterRuntimeFactory(sourceColumn, partitionColumns,
partitionComparatorFactories, inputDesc, writerFactory);
@@ -200,7 +209,7 @@
externalPrinterFactory =
new CsvExternalFilePrinterFactory(printerFactory, compressStreamFactory);
writerFactory = new ExternalFileWriterFactory(fileWriterFactory, externalPrinterFactory,
- pathResolverFactory, maxResult);
+ pathResolverFactory, maxResult, maxFileSize);
return new SinkExternalWriterRuntimeFactory(sourceColumn, partitionColumns,
partitionComparatorFactories, inputDesc, writerFactory);
} else {
@@ -234,7 +243,7 @@
(IAType) sourceType, rowGroupSize, pageSize, writerVersion);
ExternalFileWriterFactory parquetWriterFactory = new ExternalFileWriterFactory(fileWriterFactory,
- parquetPrinterFactory, pathResolverFactory, maxResult);
+ parquetPrinterFactory, pathResolverFactory, maxResult, maxFileSize);
return new SinkExternalWriterRuntimeFactory(sourceColumn, partitionColumns,
partitionComparatorFactories, inputDesc, parquetWriterFactory);
}
@@ -245,8 +254,8 @@
new ParquetExternalFilePrinterFactoryProvider(compressionCodecName, (IAType) sourceType,
rowGroupSize, pageSize, writerVersion);
return new ParquetSinkExternalWriterFactory(partitionerFactory, inputDesc, sourceColumn,
- (IAType) sourceType, maxSchemas, fileWriterFactory, maxResult, printerFactoryProvider,
- pathResolverFactory);
+ (IAType) sourceType, maxSchemas, fileWriterFactory, maxResult, maxFileSize,
+ printerFactoryProvider, pathResolverFactory);
default:
throw new CompilationException(ErrorCode.UNSUPPORTED_WRITING_FORMAT, pathSourceLocation, format,
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
index e8bae00..1ed0d31 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
@@ -28,13 +28,24 @@
private final IPathResolver pathResolver;
private final IExternalFileWriter writer;
private final int maxResultPerFile;
+ private final long maxFileSize;
+ private final boolean fileSizeBounded;
private String partitionPath;
private int tupleCounter;
- public ExternalFileWriter(IPathResolver pathResolver, IExternalFileWriter writer, int maxResultPerFile) {
+ /**
+ * @param pathResolver resolves partition directories and file names
+ * @param writer the file writer to delegate to
+ * @param maxResultPerFile maximum objects per file
+ * @param maxFileSize maximum size in bytes of a single file, or any non-positive value for no limit
+ */
+ public ExternalFileWriter(IPathResolver pathResolver, IExternalFileWriter writer, int maxResultPerFile,
+ long maxFileSize) {
this.pathResolver = pathResolver;
this.writer = writer;
this.maxResultPerFile = maxResultPerFile;
+ this.maxFileSize = maxFileSize;
+ this.fileSizeBounded = maxFileSize > 0;
}
@Override
@@ -62,7 +73,7 @@
// e.g., if max is 1000, we hit tuple 1001, we will upload and create a new file, if we only have 1000
// we will stop here, and calling the close/finish will upload whatever is written. This is to avoid
// creating and uploading empty files
- if (tupleCounter >= maxResultPerFile) {
+ if (tupleCounter >= maxResultPerFile || (fileSizeBounded && writer.getBytesWritten() >= maxFileSize)) {
newFile();
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
index f85a8ab..5fd093d 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
@@ -28,20 +28,22 @@
private final IExternalFileWriterFactory writerFactory;
private final IExternalPrinterFactory printerFactory;
private final int maxResult;
+ private final long maxFileSize;
private final IPathResolverFactory pathResolverFactory;
public ExternalFileWriterFactory(IExternalFileWriterFactory writerFactory, IExternalPrinterFactory printerFactory,
- IPathResolverFactory pathResolverFactory, int maxResult) {
+ IPathResolverFactory pathResolverFactory, int maxResult, long maxFileSize) {
this.writerFactory = writerFactory;
this.printerFactory = printerFactory;
this.pathResolverFactory = pathResolverFactory;
this.maxResult = maxResult;
+ this.maxFileSize = maxFileSize;
}
@Override
public IExternalWriter createWriter(IHyracksTaskContext context) throws HyracksDataException {
IPathResolver resolver = pathResolverFactory.createResolver(context);
IExternalFileWriter writer = writerFactory.createWriter(context, printerFactory);
- return new ExternalFileWriter(resolver, writer, maxResult);
+ return new ExternalFileWriter(resolver, writer, maxResult, maxFileSize);
}
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
index f8ae81b..0f12014 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
@@ -55,6 +55,14 @@
void write(IValueReference value) throws HyracksDataException;
/**
+ * Size of the file currently being written. The value is approximate: it reflects the bytes accepted for
+ * the target after compression, and may exclude bytes that are only produced when the file is closed.
+ *
+ * @return bytes written to the current file, or 0 if no file is open
+ */
+ long getBytesWritten() throws HyracksDataException;
+
+ /**
* Run the abort sequence in case of a failure
*/
void abort() throws HyracksDataException;
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
index 38ede08..4157f17 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
@@ -47,6 +47,15 @@
void newStream(OutputStream outputStream) throws HyracksDataException;
/**
+ * Size of the current file as accounted for by the printer, including anything it still buffers
+ *
+ * @return the printer's byte count, or 0 for printers that pass everything straight to the stream
+ */
+ default long getBytesWritten() {
+ return 0;
+ }
+
+ /**
* Print the provided value
*
* @param value to print
diff --git a/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java
new file mode 100644
index 0000000..2456293
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java
@@ -0,0 +1,185 @@
+/*
+ * 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.
+ */
+package org.apache.asterix.runtime.writer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+import org.junit.Test;
+
+public class ExternalFileWriterMaxFileSizeTest {
+
+ private static final int NO_MAX_RESULT = Integer.MAX_VALUE;
+ private static final long NO_MAX_FILE_SIZE = -1L;
+ private static final String DIRECTORY = "dir/";
+
+ @Test
+ public void testSizeUnboundedByDefault() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(1024);
+ write(fileWriter, NO_MAX_RESULT, NO_MAX_FILE_SIZE, 100);
+ assertEquals(1, fileWriter.files.size());
+ }
+
+ @Test
+ public void testRollsOverOnSizeLimit() throws HyracksDataException {
+ // 10 objects of 100 bytes each with a 200 byte limit: the limit divides evenly, so each file is
+ // closed exactly on it
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, NO_MAX_RESULT, 200, 10);
+ assertEquals(List.of(2, 2, 2, 2, 2), fileWriter.files);
+ }
+
+ @Test
+ public void testFileOvershootsWhenObjectCrossesLimit() throws HyracksDataException {
+ // 100 byte objects against a 250 byte limit: the third object takes the file to 300 bytes, past the
+ // limit, because the limit can only be enforced between objects
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, NO_MAX_RESULT, 250, 9);
+ assertEquals(List.of(3, 3, 3), fileWriter.files);
+ }
+
+ @Test
+ public void testCountLimitStillWinsWhenSmaller() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(1);
+ write(fileWriter, 4, 1024, 12);
+ assertEquals(List.of(4, 4, 4), fileWriter.files);
+ }
+
+ @Test
+ public void testSizeLimitStillWinsWhenSmaller() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, 1000, 200, 9);
+ assertEquals(List.of(2, 2, 2, 2, 1), fileWriter.files);
+ }
+
+ @Test
+ public void testObjectLargerThanLimitIsWrittenAlone() throws HyracksDataException {
+ // the limit is consulted before writing, so an oversized object lands in a file of its own that
+ // overshoots the limit rather than being split or rejected
+ CountingFileWriter fileWriter = new CountingFileWriter(4096);
+ write(fileWriter, NO_MAX_RESULT, 1024, 3);
+ assertEquals(List.of(1, 1, 1), fileWriter.files);
+ }
+
+ @Test
+ public void testNoEmptyFileIsCreated() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, NO_MAX_RESULT, 200, 10);
+ for (int objectsInFile : fileWriter.files) {
+ assertTrue("a file was created without any object in it", objectsInFile > 0);
+ }
+ }
+
+ @Test
+ public void testSizeIsNotConsultedWhenUnbounded() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(1024);
+ write(fileWriter, NO_MAX_RESULT, NO_MAX_FILE_SIZE, 100);
+ assertEquals(0, fileWriter.sizeQueries);
+ }
+
+ @Test
+ public void testSizeIsConsultedForEveryObject() throws HyracksDataException {
+ // the limit is enforced as tightly as it can be: the size is read once per object, so the file can
+ // only ever overshoot by the object that crosses the limit
+ CountingFileWriter fileWriter = new CountingFileWriter(1);
+ int objects = 10000;
+ write(fileWriter, NO_MAX_RESULT, 1024 * 1024, objects);
+ assertEquals(1, fileWriter.files.size());
+ assertEquals(objects, fileWriter.sizeQueries);
+ }
+
+ private static void write(CountingFileWriter fileWriter, int maxResultPerFile, long maxFileSize, int objects)
+ throws HyracksDataException {
+ ExternalFileWriter writer =
+ new ExternalFileWriter(new FixedPathResolver(), fileWriter, maxResultPerFile, maxFileSize);
+ writer.open();
+ writer.initNewPartition(null);
+ for (int i = 0; i < objects; i++) {
+ writer.write(new VoidPointable());
+ }
+ writer.close();
+ }
+
+ private static class FixedPathResolver implements IPathResolver {
+ private int fileCounter;
+
+ @Override
+ public String getPartitionDirectory(IFrameTupleReference tuple) {
+ return DIRECTORY;
+ }
+
+ @Override
+ public String getNextFileName() {
+ return "file-" + fileCounter++;
+ }
+ }
+
+ /**
+ * Reports a fixed number of bytes per written object and records how many objects landed in each file.
+ */
+ private static class CountingFileWriter implements IExternalFileWriter {
+ private final List<Integer> files = new ArrayList<>();
+ private final int bytesPerObject;
+ private int sizeQueries;
+
+ CountingFileWriter(int bytesPerObject) {
+ this.bytesPerObject = bytesPerObject;
+ }
+
+ @Override
+ public void open() {
+ }
+
+ @Override
+ public void validate(String directory) {
+ }
+
+ @Override
+ public boolean newFile(String directory, String fileName) {
+ files.add(0);
+ return true;
+ }
+
+ @Override
+ public void write(IValueReference value) {
+ files.set(files.size() - 1, files.get(files.size() - 1) + 1);
+ }
+
+ @Override
+ public long getBytesWritten() {
+ sizeQueries++;
+ return (long) files.get(files.size() - 1) * bytesPerObject;
+ }
+
+ @Override
+ public void abort() {
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupWriteOperatorNodePushable.java b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupWriteOperatorNodePushable.java
index 34ae038..a4e4674 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupWriteOperatorNodePushable.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/group/external/ExternalGroupWriteOperatorNodePushable.java
@@ -174,7 +174,8 @@
ExternalHashGroupBy groupBy = new ExternalHashGroupBy(this, table, runFileWriters, partialAggRecordDesc);
reader.open();
try {
- VSizeFrame frame = ctx.allocateVSizeFrame();
+ // read back by RunFileReader, which requires a frame that shrinks on reset
+ VSizeFrame frame = new VSizeFrame(ctx);
while (reader.nextFrame(frame)) {
groupBy.insert(frame.getBuffer());
}
diff --git a/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/RetainingFrameOptimizedHybridHashJoinTest.java b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/RetainingFrameOptimizedHybridHashJoinTest.java
new file mode 100644
index 0000000..ded8e33
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-examples/hyracks-integration-tests/src/test/java/org/apache/hyracks/tests/integration/RetainingFrameOptimizedHybridHashJoinTest.java
@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+package org.apache.hyracks.tests.integration;
+
+import org.apache.hyracks.api.comm.VSizeFrame;
+import org.apache.hyracks.api.context.IHyracksFrameMgrContext;
+import org.apache.hyracks.api.context.IHyracksTaskContext;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.test.support.TestUtils;
+import org.junit.Before;
+
+/**
+ * Runs the {@link OptimizedHybridHashJoinTest} cases against a frame factory whose frames retain their expanded
+ * allocation across {@link VSizeFrame#reset()}, as the configurable frame policies do. Frames read through
+ * {@code RunFileReader} must not come from the factory, so the join's results and its memory reservation invariants
+ * are expected to be unaffected by the retention.
+ */
+public class RetainingFrameOptimizedHybridHashJoinTest extends OptimizedHybridHashJoinTest {
+
+ @Before
+ public void useRetainingFrames() {
+ IHyracksTaskContext taskCtx = TestUtils.create(frameSize);
+ taskCtx.getJobletContext().getServiceContext().setVSizeFrameFactoryProvider(kind -> RetainingVSizeFrame::new);
+ ctx = taskCtx.getJobletContext();
+ }
+
+ private static class RetainingVSizeFrame extends VSizeFrame {
+
+ RetainingVSizeFrame(IHyracksFrameMgrContext ctx) throws HyracksDataException {
+ super(ctx);
+ }
+
+ @Override
+ public void reset() {
+ buffer.clear();
+ }
+ }
+}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21493?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Ib409d6c9022f36a94946796aa22d772028a35659
Gerrit-Change-Number: 21493
Gerrit-PatchSet: 1
Gerrit-Owner: Rithwik Koul <ri...@couchbase.com>