>From Hussain Towaileb <[email protected]>:
Hussain Towaileb has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21403?usp=email )
Change subject: [NO ISSUE][EXT]: Handle interrupts when reading external
collections
......................................................................
[NO ISSUE][EXT]: Handle interrupts when reading external collections
Details:
- handle interrupts when reading from external collections.
- handle invalid utf8 chars when reading from external collections.
- add property to handle invalid characters, values are:
- replace (default) to replace invalid chars with ? and continue reading
- fail to fail fast and stop reading
Ext-ref: MB-72470, MB-72724
Change-Id: Ib4c6f48470b77d3ba1e7bcc35240df468835f132
---
A asterixdb/asterix-app/data/json/illegal-character.json
M
asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.000.container.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.001.ddl.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.002.query.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.099.ddl.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.000.container.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.001.ddl.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.002.query.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.099.ddl.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/illegal-character/replace/result.002.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-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/blob/AzureBlobInputStream.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/datalake/AzureDataLakeInputStream.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
M
asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
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/ExternalDataUtils.java
A
asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/stream/AsterixInputStreamReaderIllegalCharacterTest.java
M
hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/ExceptionUtils.java
M
hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/annotations/AiProvenance.java
22 files changed, 445 insertions(+), 7 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/03/21403/1
diff --git a/asterixdb/asterix-app/data/json/illegal-character.json
b/asterixdb/asterix-app/data/json/illegal-character.json
new file mode 100644
index 0000000..f2dbfc1
--- /dev/null
+++ b/asterixdb/asterix-app/data/json/illegal-character.json
@@ -0,0 +1 @@
+{"id": 1, "value": "helloéworld"}
diff --git
a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
index 8a94453..e4e6c70 100644
---
a/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
+++
b/asterixdb/asterix-app/src/test/java/org/apache/asterix/test/external_dataset/ExternalDatasetTestUtils.java
@@ -45,6 +45,7 @@
import org.apache.commons.io.FilenameUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hyracks.api.util.IoUtil;
+import org.apache.hyracks.util.annotations.AiProvenance;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.Assert;
@@ -75,6 +76,7 @@
public static final String PARQUET_DEFINITION = "parquet-data/reviews/";
public static final String PARQUET_NULL_TEST_DIRECTORY =
"parquet-data/null-test/";
public static final String AVRO_DEFINITION = "avro-data/reviews/";
+ public static final String ILLEGAL_CHARACTER_DEFINITION =
"illegal-character-data/";
// This is used for a test to generate over 1000 number of files
public static final String OVER_1000_OBJECTS_PATH = "over-1000-objects";
@@ -210,9 +212,21 @@
loadDeltaTableFiles();
LOGGER.info("Delta files added successfully");
+ LOGGER.info("Adding illegal-character file to the bucket");
+ loadIllegalCharacterFile();
+ LOGGER.info("Illegal-character file added successfully");
+
LOGGER.info("Files added successfully");
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_5, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.TEST_GENERATED, notes = "Uploads a file with a
single illegal (invalid UTF-8) byte, used to test the "
+ + "illegalCharacterHandling property (replace vs fail)")
+ private static void loadIllegalCharacterFile() {
+ String fileName = "illegal-character.json";
+ Path filePath = Paths.get(JSON_DATA_PATH, fileName);
+ playgroundDataLoader.upload(ILLEGAL_CHARACTER_DEFINITION + fileName,
filePath.toString(), true, false);
+ }
+
public static void prepareBrowseContainer() {
/*
file hierarchy inside browse container
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.000.container.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.000.container.sqlpp
new file mode 100644
index 0000000..068dbd3
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.000.container.sqlpp
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+// create container with data
+illegal-character-fail illegal-character-data data/json/illegal-character.json
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.001.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.001.ddl.sqlpp
new file mode 100644
index 0000000..6fe8692
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.001.ddl.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.
+ */
+
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+USE test;
+
+DROP TYPE t1 IF EXISTS;
+CREATE TYPE t1 AS {};
+
+DROP DATASET ds1 IF EXISTS;
+CREATE EXTERNAL DATASET ds1(t1) USING %adapter% (
+%template%,
+%additional_Properties%
+("container"="illegal-character-fail"),
+("definition"="illegal-character-data"),
+("format"="json"),
+("illegalCharacterHandling"="fail")
+);
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.002.query.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.002.query.sqlpp
new file mode 100644
index 0000000..3dedd2f
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.002.query.sqlpp
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+USE test;
+
+FROM ds1 v SELECT VALUE v;
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.099.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.099.ddl.sqlpp
new file mode 100644
index 0000000..36b2bab
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/fail/test.099.ddl.sqlpp
@@ -0,0 +1,20 @@
+/*
+ * 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;
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.000.container.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.000.container.sqlpp
new file mode 100644
index 0000000..e8daf0e
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.000.container.sqlpp
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+// create container with data
+illegal-character-replace illegal-character-data
data/json/illegal-character.json
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.001.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.001.ddl.sqlpp
new file mode 100644
index 0000000..5e5ecfe
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.001.ddl.sqlpp
@@ -0,0 +1,34 @@
+/*
+ * 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;
+
+DROP TYPE t1 IF EXISTS;
+CREATE TYPE t1 AS {};
+
+DROP DATASET ds1 IF EXISTS;
+CREATE EXTERNAL DATASET ds1(t1) USING %adapter% (
+%template%,
+%additional_Properties%
+("container"="illegal-character-replace"),
+("definition"="illegal-character-data"),
+("format"="json")
+);
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.002.query.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.002.query.sqlpp
new file mode 100644
index 0000000..3dedd2f
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.002.query.sqlpp
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+USE test;
+
+FROM ds1 v SELECT VALUE v;
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.099.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.099.ddl.sqlpp
new file mode 100644
index 0000000..36b2bab
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/external-dataset/common/illegal-character/replace/test.099.ddl.sqlpp
@@ -0,0 +1,20 @@
+/*
+ * 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;
\ No newline at end of file
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/illegal-character/replace/result.002.adm
b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/illegal-character/replace/result.002.adm
new file mode 100644
index 0000000..8cab9cf
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/results/external-dataset/common/illegal-character/replace/result.002.adm
@@ -0,0 +1 @@
+{ "id": 1, "value": "hello�world" }
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 993372e..0e5ce4d 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
@@ -550,6 +550,19 @@
</compilation-unit>
</test-case>
<test-case FilePath="external-dataset">
+ <compilation-unit name="common/illegal-character/replace">
+ <placeholder name="adapter" value="AZUREBLOB" />
+ <output-dir
compare="Text">common/illegal-character/replace</output-dir>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="external-dataset">
+ <compilation-unit name="common/illegal-character/fail">
+ <placeholder name="adapter" value="AZUREBLOB" />
+ <output-dir compare="Text">common/illegal-character/fail</output-dir>
+ <expected-error>External source error</expected-error>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="external-dataset">
<compilation-unit name="common/definition-does-not-exist">
<placeholder name="adapter" value="AZUREBLOB" />
<output-dir
compare="Text">common/definition-does-not-exist</output-dir>
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 d19d5fc..28da731 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
@@ -1224,6 +1224,19 @@
</compilation-unit>
</test-case>
<test-case FilePath="external-dataset">
+ <compilation-unit name="common/illegal-character/replace">
+ <placeholder name="adapter" value="S3" />
+ <output-dir
compare="Text">common/illegal-character/replace</output-dir>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="external-dataset">
+ <compilation-unit name="common/illegal-character/fail">
+ <placeholder name="adapter" value="S3" />
+ <output-dir compare="Text">common/illegal-character/fail</output-dir>
+ <expected-error>External source error</expected-error>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="external-dataset">
<compilation-unit name="common/definition-does-not-exist">
<placeholder name="adapter" value="S3" />
<output-dir
compare="Text">common/definition-does-not-exist</output-dir>
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/blob/AzureBlobInputStream.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/blob/AzureBlobInputStream.java
index 59e07b6..9d2e0ad 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/blob/AzureBlobInputStream.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/blob/AzureBlobInputStream.java
@@ -93,4 +93,14 @@
throw HyracksDataException.create(ex);
}
}
+
+ @Override
+ public boolean stop() {
+ try {
+ close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ return false;
+ }
}
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/datalake/AzureDataLakeInputStream.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/datalake/AzureDataLakeInputStream.java
index 8ba3e79..cf521bb 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/datalake/AzureDataLakeInputStream.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/azure/datalake/AzureDataLakeInputStream.java
@@ -93,4 +93,14 @@
throw HyracksDataException.create(ex);
}
}
+
+ @Override
+ public boolean stop() {
+ try {
+ close();
+ } catch (IOException e) {
+ // Ignore
+ }
+ return false;
+ }
}
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
index 5ef63bc..15ac0b9 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/record/reader/stream/StreamRecordReader.java
@@ -43,7 +43,8 @@
protected void configure(AsterixInputStream inputStream, Map<String,
String> config) {
int bufferSize = ExternalDataUtils.getOrDefaultBufferSize(config);
- this.reader = new AsterixInputStreamReader(inputStream, bufferSize);
+ boolean failOnIllegalCharacter =
ExternalDataUtils.shouldFailOnIllegalCharacter(config);
+ this.reader = new AsterixInputStreamReader(inputStream, bufferSize,
failOnIllegalCharacter);
record = new CharArrayRecord();
inputBuffer = new char[bufferSize];
setSuppliers(config, reader::getStreamName,
reader::getPreviousStreamName);
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
index 2dc323e..72ae5e9 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/input/stream/AsterixInputStreamReader.java
@@ -18,21 +18,36 @@
*/
package org.apache.asterix.external.input.stream;
+import static org.apache.hyracks.api.util.ExceptionUtils.getMessageOrToString;
+
import java.io.IOException;
+import java.io.InterruptedIOException;
import java.io.Reader;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
+import java.nio.charset.CharacterCodingException;
import java.nio.charset.CharsetDecoder;
+import java.nio.charset.CoderResult;
+import java.nio.charset.CodingErrorAction;
import java.nio.charset.StandardCharsets;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.common.exceptions.RuntimeDataException;
import org.apache.asterix.external.api.AsterixInputStream;
import org.apache.asterix.external.dataflow.AbstractFeedDataFlowController;
import org.apache.asterix.external.util.IFeedLogManager;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.util.annotations.AiProvenance;
public class AsterixInputStreamReader extends Reader {
+
+ // Unicode REPLACEMENT CHARACTER (U+FFFD) -- the standard substitute for
an illegal/unmappable
+ // byte sequence during decoding (e.g. WHATWG Encoding Standard, Python's
errors='replace').
+ private static final String ILLEGAL_CHARACTER_REPLACEMENT = "\uFFFD";
+
private final AsterixInputStream in;
private final CharsetDecoder decoder;
+ private final boolean failOnIllegalCharacter;
private byte[] bytes;
protected final ByteBuffer byteBuffer;
protected final CharBuffer charBuffer;
@@ -40,14 +55,39 @@
protected boolean remaining = false;
public AsterixInputStreamReader(AsterixInputStream in, int bufferSize) {
+ this(in, bufferSize, false);
+ }
+
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_5, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Illegal (invalid UTF-8)
characters previously stalled the decoder forever since its "
+ + "CoderResult was never consulted; default to substituting U+FFFD
and continuing, with an "
+ + "explicit fail-fast option that reports the error immediately
instead of stalling")
+ public AsterixInputStreamReader(AsterixInputStream in, int bufferSize,
boolean failOnIllegalCharacter) {
this.in = in;
+ this.failOnIllegalCharacter = failOnIllegalCharacter;
this.decoder = StandardCharsets.UTF_8.newDecoder();
+ if (!failOnIllegalCharacter) {
+
decoder.onMalformedInput(CodingErrorAction.REPLACE).onUnmappableCharacter(CodingErrorAction.REPLACE)
+ .replaceWith(ILLEGAL_CHARACTER_REPLACEMENT);
+ }
this.bytes = new byte[bufferSize];
this.byteBuffer = ByteBuffer.wrap(bytes);
this.charBuffer = CharBuffer.allocate(bufferSize);
this.byteBuffer.flip();
}
+ @AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_5, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.GENERATED, notes = "Consult the decoder's
CoderResult (previously discarded) and, in fail-fast mode, report an "
+ + "illegal character as EXTERNAL_SOURCE_ERROR instead of the
generic untyped wrapping")
+ private void decode(boolean endOfInput) throws IOException {
+ CoderResult result = decoder.decode(byteBuffer, charBuffer,
endOfInput);
+ if (failOnIllegalCharacter && result.isError()) {
+ try {
+ result.throwException();
+ } catch (CharacterCodingException e) {
+ throw
RuntimeDataException.create(ErrorCode.EXTERNAL_SOURCE_ERROR, e,
getMessageOrToString(e));
+ }
+ }
+ }
+
public void stop() throws IOException {
try {
in.stop();
@@ -79,7 +119,7 @@
while (charBuffer.position() == 0) {
if (byteBuffer.hasRemaining()) {
remaining = true;
- decoder.decode(byteBuffer, charBuffer, false);
+ decode(false);
System.arraycopy(charBuffer.array(), 0, cbuf, offset,
charBuffer.position());
if (charBuffer.position() > 0) {
return charBuffer.position();
@@ -90,12 +130,14 @@
byteBuffer.position(byteBuffer.remaining());
byteBuffer.limit(byteBuffer.capacity()); //set limit to
capacity for the new bytes
while (len == 0) {
+ checkInterrupted();
len = in.read(bytes, byteBuffer.position(),
bytes.length - byteBuffer.position());
}
}
} else {
byteBuffer.clear();
while (len == 0) {
+ checkInterrupted();
len = in.read(bytes, 0, bytes.length);
}
}
@@ -110,12 +152,20 @@
}
byteBuffer.flip();
remaining = false;
- decoder.decode(byteBuffer, charBuffer, false);
+ decode(false);
System.arraycopy(charBuffer.array(), 0, cbuf, offset,
charBuffer.position());
}
return charBuffer.position();
}
+ @AiProvenance(agent = AiProvenance.Agent.DEEPSEEK_CODER, tool =
AiProvenance.Tool.FACTORY_CLI, contributionKind =
AiProvenance.ContributionKind.ASSISTED, notes = "Check for thread interruption
(re-interrupts thread and throws InterruptedIOException) to prevent infinite
spin in read loop when stream is cancelled")
+ private static void checkInterrupted() throws IOException {
+ if (Thread.interrupted()) {
+ Thread.currentThread().interrupt();
+ throw new InterruptedIOException("Thread interrupted while reading
stream");
+ }
+ }
+
@Override
public void close() throws IOException {
in.close();
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 8f0d363..fbdfb91 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
@@ -47,6 +47,10 @@
public static final String KEY_DATASET_DATAVERSE = "dataset-dataverse";
// used to specify the socket addresses when reading data from sockets
public static final String KEY_SOCKETS = "sockets";
+ // specify how an illegal (invalid UTF-8) character in the source should
be handled
+ public static final String KEY_ILLEGAL_CHARACTER_HANDLING =
"illegalCharacterHandling";
+ // fail the read as soon as an illegal character is encountered
+ public static final String ILLEGAL_CHARACTER_HANDLING_FAIL = "fail";
// specify whether the socket address points to an NC or an IP
public static final String KEY_MODE = "address-type";
// specify the HDFS name node address when reading HDFS data
diff --git
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
index 713a1fb..b405c88 100644
---
a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
+++
b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataUtils.java
@@ -31,10 +31,12 @@
import static org.apache.asterix.common.utils.CSVConstants.KEY_QUOTE;
import static
org.apache.asterix.external.util.ExternalDataConstants.DEFINITION_FIELD_NAME;
import static
org.apache.asterix.external.util.ExternalDataConstants.DISABLE_SSL_VERIFY_FIELD_NAME;
+import static
org.apache.asterix.external.util.ExternalDataConstants.ILLEGAL_CHARACTER_HANDLING_FAIL;
import static
org.apache.asterix.external.util.ExternalDataConstants.KEY_ADAPTER_NAME_AZURE_DATALAKE;
import static
org.apache.asterix.external.util.ExternalDataConstants.KEY_ADAPTER_NAME_AZURE_DATALAKE_ALIAS;
import static
org.apache.asterix.external.util.ExternalDataConstants.KEY_EXCLUDE;
import static
org.apache.asterix.external.util.ExternalDataConstants.KEY_EXTERNAL_SCAN_BUFFER_SIZE;
+import static
org.apache.asterix.external.util.ExternalDataConstants.KEY_ILLEGAL_CHARACTER_HANDLING;
import static
org.apache.asterix.external.util.ExternalDataConstants.KEY_INCLUDE;
import static org.apache.asterix.external.util.ExternalDataConstants.KEY_PATH;
import static
org.apache.asterix.external.util.ExternalDataConstants.KEY_RECORD_END;
@@ -173,6 +175,12 @@
return bufferSize != null ? Integer.parseInt(bufferSize) :
ExternalDataConstants.DEFAULT_BUFFER_SIZE;
}
+ // Defaults to replace-and-continue; only "fail" opts into failing fast on
an illegal character.
+ public static boolean shouldFailOnIllegalCharacter(Map<String, String>
configuration) {
+ String mode = configuration.get(KEY_ILLEGAL_CHARACTER_HANDLING);
+ return ILLEGAL_CHARACTER_HANDLING_FAIL.equalsIgnoreCase(mode);
+ }
+
// Get a delimiter from the given configuration
public static char validateGetDelimiter(Map<String, String> configuration)
throws HyracksDataException {
return validateCharOrDefault(configuration, KEY_DELIMITER,
ExternalDataConstants.DEFAULT_DELIMITER.charAt(0));
diff --git
a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/stream/AsterixInputStreamReaderIllegalCharacterTest.java
b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/stream/AsterixInputStreamReaderIllegalCharacterTest.java
new file mode 100644
index 0000000..b60c47a
--- /dev/null
+++
b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/input/stream/AsterixInputStreamReaderIllegalCharacterTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.input.stream;
+
+import java.io.IOException;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.apache.asterix.external.util.ExternalDataConstants;
+import org.apache.asterix.external.util.FileSystemWatcher;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.util.annotations.AiProvenance;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Regression test for AsterixInputStreamReader's handling of an illegal
+ * (invalid UTF-8) character byte in the source: previously, CharsetDecoder's
+ * CoderResult was never consulted, so the decoder would permanently stall at
+ * the bad byte -- never skipping or reporting past it -- until the internal
+ * byte buffer saturated and every subsequent read became a non-blocking,
+ * CPU-pegging infinite spin. The fix configures the decoder to substitute
+ * illegal characters with the Unicode REPLACEMENT CHARACTER (U+FFFD) and
+ * keep reading by default, with an explicit fail-fast option that reports
+ * the error immediately as an EXTERNAL_SOURCE_ERROR instead of stalling.
+ */
+@AiProvenance(agent = AiProvenance.Agent.CLAUDE_SONNET_5, tool =
AiProvenance.Tool.CLAUDE_CODE_UI, contributionKind =
AiProvenance.ContributionKind.TEST_GENERATED, notes = "Regression test for
illegal-character handling (replace-by-default, fail-fast option) in "
+ + "AsterixInputStreamReader")
+public class AsterixInputStreamReaderIllegalCharacterTest {
+
+ private Path tempFile;
+
+ @After
+ public void cleanup() throws IOException {
+ if (tempFile != null) {
+ Files.deleteIfExists(tempFile);
+ }
+ }
+
+ private Path writeIllegalCharacterFile() throws IOException {
+ Path file = Files.createTempFile("illegal-character-utf8", ".json");
+ byte[] prefix = "{\"value\": \"".getBytes(StandardCharsets.US_ASCII);
+ byte[] illegalCharacter = { (byte) 0xE9 };
+ byte[] suffix = "hello\"}\n".getBytes(StandardCharsets.US_ASCII);
+ try (java.io.OutputStream out = Files.newOutputStream(file)) {
+ out.write(prefix);
+ out.write(illegalCharacter);
+ out.write(suffix);
+ }
+ return file;
+ }
+
+ private AsterixInputStreamReader createReader(Path file, boolean
failOnIllegalCharacter) throws Exception {
+ FileSystemWatcher watcher = new
FileSystemWatcher(Collections.singletonList(file), null, false);
+ LocalFSInputStream in = new LocalFSInputStream(watcher);
+ return new AsterixInputStreamReader(in,
ExternalDataConstants.DEFAULT_BUFFER_SIZE, failOnIllegalCharacter);
+ }
+
+ @Test(timeout = 15000)
+ public void testDefaultReplacesIllegalCharacterAndContinues() throws
Exception {
+ tempFile = writeIllegalCharacterFile();
+ AsterixInputStreamReader reader = createReader(tempFile, false);
+
+ char[] buf = new char[ExternalDataConstants.DEFAULT_BUFFER_SIZE];
+ int len = reader.read(buf);
+
+ Assert.assertEquals("{\"value\": \"\uFFFDhello\"}\n", new String(buf,
0, len));
+ }
+
+ @Test(timeout = 15000)
+ public void testFailOnIllegalCharacterThrowsImmediately() throws Exception
{
+ tempFile = writeIllegalCharacterFile();
+ AsterixInputStreamReader reader = createReader(tempFile, true);
+
+ try {
+ reader.read(new char[ExternalDataConstants.DEFAULT_BUFFER_SIZE]);
+ Assert.fail("expected an EXTERNAL_SOURCE_ERROR for the illegal
character");
+ } catch (HyracksDataException e) {
+ // reported immediately as a typed EXTERNAL_SOURCE_ERROR, not
stalled
+ Assert.assertTrue("expected EXTERNAL_SOURCE_ERROR, got: " + e,
e.matches(ErrorCode.EXTERNAL_SOURCE_ERROR));
+ Assert.assertTrue("expected the cause to be a
CharacterCodingException, got: " + e.getCause(),
+ e.getCause() instanceof CharacterCodingException);
+ }
+ }
+
+ @Test
+ public void testWellFormedFileReadsNormally() throws Exception {
+ tempFile = Files.createTempFile("well-formed-utf8", ".json");
+ String content = "{\"value\": \"hello world\"}\n";
+ Files.write(tempFile, content.getBytes(StandardCharsets.UTF_8));
+
+ AsterixInputStreamReader reader = createReader(tempFile, false);
+ char[] buf = new char[ExternalDataConstants.DEFAULT_BUFFER_SIZE];
+ int len = reader.read(buf);
+
+ Assert.assertEquals(content, new String(buf, 0, len));
+ }
+}
diff --git
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/ExceptionUtils.java
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/ExceptionUtils.java
index 81de6ed..17f7827 100644
---
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/ExceptionUtils.java
+++
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/util/ExceptionUtils.java
@@ -18,6 +18,7 @@
*/
package org.apache.hyracks.api.util;
+import java.io.InterruptedIOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -86,7 +87,7 @@
private static boolean possibleRootCause(Throwable exception) {
Throwable cause = exception;
while ((cause = cause.getCause()) != null) {
- if (cause instanceof java.lang.InterruptedException
+ if (cause instanceof java.lang.InterruptedException || cause
instanceof InterruptedIOException
|| cause instanceof
java.nio.channels.ClosedChannelException) {
return false;
}
@@ -105,7 +106,7 @@
* @return the root exception, or null if both parameters are null
*/
public static <T extends Throwable> T suppress(T first, T second) {
- if (second instanceof InterruptedException) {
+ if (second instanceof InterruptedException || second instanceof
InterruptedIOException) {
Thread.currentThread().interrupt();
}
if (first == null) {
@@ -149,10 +150,11 @@
}
public static boolean causedByInterrupt(Throwable th, boolean
skipInterruptedCheck) {
- if (th instanceof InterruptedException) {
+ if (th instanceof InterruptedException || th instanceof
InterruptedIOException) {
return true;
}
- boolean isCausedByInterrupt = getRootCause(th) instanceof
InterruptedException;
+ boolean isCausedByInterrupt =
+ getRootCause(th) instanceof InterruptedException ||
getRootCause(th) instanceof InterruptedIOException;
if (!skipInterruptedCheck && isCausedByInterrupt &&
!Thread.currentThread().isInterrupted()) {
LOGGER.warn("InterruptedException suppressed and
!Thread.currentThread().isInterrupted()", th);
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/annotations/AiProvenance.java
b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/annotations/AiProvenance.java
index f20956d..2afaafd 100644
---
a/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/annotations/AiProvenance.java
+++
b/hyracks-fullstack/hyracks/hyracks-util/src/main/java/org/apache/hyracks/util/annotations/AiProvenance.java
@@ -203,6 +203,7 @@
// =========================
// Anthropic â Claude Sonnet
// =========================
+ CLAUDE_SONNET_5(ANTHROPIC, "claude-sonnet-5", "Claude Sonnet 5"),
CLAUDE_SONNET_4(ANTHROPIC, "claude-4-sonnet", "Claude Sonnet 4"),
CLAUDE_SONNET_4_5(ANTHROPIC, "claude-4-sonnet-4.5", "Claude Sonnet
4.5"),
CLAUDE_SONNET_4_6(ANTHROPIC, "claude-4-sonnet-4.6", "Claude Sonnet
4.6"),
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21403?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: Ib4c6f48470b77d3ba1e7bcc35240df468835f132
Gerrit-Change-Number: 21403
Gerrit-PatchSet: 1
Gerrit-Owner: Hussain Towaileb <[email protected]>
