You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by jd...@apache.org on 2018/03/12 22:29:28 UTC
hive git commit: HIVE-18907: Create utility to fix acid key index
issue from HIVE-18817 (Jason Dere, reviewed by Prasanth Jayachandran)
Repository: hive
Updated Branches:
refs/heads/master 1629ec058 -> d717d3853
HIVE-18907: Create utility to fix acid key index issue from HIVE-18817 (Jason Dere, reviewed by Prasanth Jayachandran)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/d717d385
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/d717d385
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/d717d385
Branch: refs/heads/master
Commit: d717d3853469afc95dbb46a05f52e1b4c7304a89
Parents: 1629ec0
Author: Jason Dere <jd...@hortonworks.com>
Authored: Mon Mar 12 15:28:50 2018 -0700
Committer: Jason Dere <jd...@hortonworks.com>
Committed: Mon Mar 12 15:28:50 2018 -0700
----------------------------------------------------------------------
bin/ext/fixacidkeyindex.sh | 32 ++
.../hadoop/hive/ql/io/orc/FixAcidKeyIndex.java | 374 +++++++++++++++++++
.../hive/ql/io/orc/TestFixAcidKeyIndex.java | 301 +++++++++++++++
3 files changed, 707 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hive/blob/d717d385/bin/ext/fixacidkeyindex.sh
----------------------------------------------------------------------
diff --git a/bin/ext/fixacidkeyindex.sh b/bin/ext/fixacidkeyindex.sh
new file mode 100644
index 0000000..28af2ea
--- /dev/null
+++ b/bin/ext/fixacidkeyindex.sh
@@ -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.
+
+THISSERVICE=fixacidkeyindex
+export SERVICE_LIST="${SERVICE_LIST}${THISSERVICE} "
+
+fixacidkeyindex () {
+ CLASS=org.apache.hadoop.hive.ql.io.orc.FixAcidKeyIndex
+ HIVE_OPTS=''
+ execHiveCmd $CLASS "$@"
+}
+
+fixacidkeyindex_help () {
+ echo "usage ./hive fixacidkeyindex [-h] --check-only|--recover [--backup-path <new-path>] <path_to_orc_file_or_directory>"
+ echo ""
+ echo " --check-only Check acid orc file for valid acid key index and exit without fixing"
+ echo " --recover Fix the acid key index for acid orc file if it requires fixing"
+ echo " --backup-path <new_path> Specify a backup path to store the corrupted files (default: /tmp)"
+ echo " --help (-h) Print help message"
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/d717d385/ql/src/java/org/apache/hadoop/hive/ql/io/orc/FixAcidKeyIndex.java
----------------------------------------------------------------------
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/orc/FixAcidKeyIndex.java b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/FixAcidKeyIndex.java
new file mode 100644
index 0000000..6920938
--- /dev/null
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/orc/FixAcidKeyIndex.java
@@ -0,0 +1,374 @@
+/*
+ * 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.hadoop.hive.ql.io.orc;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.OptionBuilder;
+import org.apache.commons.cli.Options;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.PathFilter;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.io.RecordIdentifier;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.IntObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.LongObjectInspector;
+import org.apache.orc.CompressionKind;
+import org.apache.orc.StripeInformation;
+import org.apache.orc.OrcProto.StripeStatistics;
+import org.apache.orc.impl.OrcAcidUtils;
+import org.apache.orc.tools.FileDump;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Utility to check and fix the ACID key index of an ORC file if it has been written incorrectly
+ * due to HIVE-18817.
+ * The condition that will be checked in the ORC file will be if the number of stripes in the
+ * acid key index matches the number of stripes in the ORC StripeInformation.
+ */
+public class FixAcidKeyIndex {
+ public final static Logger LOG = LoggerFactory.getLogger(FixAcidKeyIndex.class);
+
+ public static final String DEFAULT_BACKUP_PATH = System.getProperty("java.io.tmpdir");
+ private static final Charset UTF8 = Charset.forName("UTF-8");
+ private static final CharsetDecoder utf8Decoder = UTF8.newDecoder();
+
+ public static void main(String[] args) throws Exception {
+ Configuration conf = new Configuration();
+
+ Options opts = createOptions();
+ CommandLine cli = new GnuParser().parse(opts, args);
+
+ if (cli.hasOption('h')) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("fixacidkeyindex", opts);
+ return;
+ }
+
+ String backupPath = DEFAULT_BACKUP_PATH;
+ if (cli.hasOption("backup-path")) {
+ backupPath = cli.getOptionValue("backup-path");
+ }
+
+ boolean checkOnly = cli.hasOption("check-only");
+ boolean recover = cli.hasOption("recover");
+
+ String[] files = cli.getArgs();
+ if (files.length == 0) {
+ System.err.println("Error : ORC files are not specified");
+ return;
+ }
+
+ // if the specified path is directory, iterate through all files
+ List<String> filesInPath = new ArrayList<>();
+ for (String filename : files) {
+ Path path = new Path(filename);
+ filesInPath.addAll(getAllFilesInPath(path, conf));
+ }
+
+ if (checkOnly) {
+ checkFiles(conf, filesInPath);
+ } else if (recover) {
+ recoverFiles(conf, filesInPath, backupPath);
+ } else {
+ System.err.println("check-only or recover option must be specified");
+ }
+ }
+
+ static boolean isAcidKeyIndexValid(Reader reader) {
+ // The number of stripes should match the key index count
+ List<StripeInformation> stripes = reader.getStripes();
+ RecordIdentifier[] keyIndex = OrcRecordUpdater.parseKeyIndex(reader);
+ if (keyIndex == null) {
+ return false;
+ }
+
+ for (int idx = 0; idx < keyIndex.length; ++idx) {
+ if (keyIndex[idx] == null) {
+ LOG.info("*** keyIndex[" + idx + "] is null");
+ return false;
+ }
+ }
+
+ return stripes.size() == keyIndex.length;
+ }
+
+ static void recoverFiles(Configuration conf, List<String> fileList, String backup) {
+ for (String fileName : fileList) {
+ try {
+ Path filePath = new Path(fileName);
+ recoverFile(conf, filePath, backup);
+ } catch (Exception err) {
+ System.err.println("ERROR recovering " + fileName);
+ err.printStackTrace(System.err);
+ }
+ }
+ }
+
+ static void checkFiles(Configuration conf, List<String> fileList) {
+ for (String fileName : fileList) {
+ try {
+ Path filePath = new Path(fileName);
+ checkFile(conf, filePath);
+ } catch (Exception err) {
+ System.err.println("ERROR checking " + fileName);
+ err.printStackTrace(System.err);
+ }
+ }
+ }
+
+ static void checkFile(Configuration conf, Path inputPath) throws IOException {
+ FileSystem fs = inputPath.getFileSystem(conf);
+ Reader reader = OrcFile.createReader(fs, inputPath);
+
+ if (OrcInputFormat.isOriginal(reader)) {
+ System.out.println(inputPath + " is not an acid file");
+ return;
+ }
+
+ boolean validIndex = isAcidKeyIndexValid(reader);
+ System.out.println("Checking " + inputPath + " - acid key index is " +
+ (validIndex ? "valid" : "invalid"));
+ }
+
+ static void recoverFile(Configuration conf, Path inputPath, String backup) throws IOException {
+ FileSystem fs = inputPath.getFileSystem(conf);
+ Reader reader = OrcFile.createReader(fs, inputPath);
+
+ if (OrcInputFormat.isOriginal(reader)) {
+ System.out.println(inputPath + " is not an acid file. No need to recover.");
+ return;
+ }
+
+ boolean validIndex = isAcidKeyIndexValid(reader);
+ if (validIndex) {
+ System.out.println(inputPath + " has a valid acid key index. No need to recover.");
+ return;
+ }
+
+ System.out.println("Recovering " + inputPath);
+
+ Path recoveredPath = getRecoveryFile(inputPath);
+ // make sure that file does not exist
+ if (fs.exists(recoveredPath)) {
+ fs.delete(recoveredPath, false);
+ }
+
+ // Writer should match the orc configuration from the original file
+ OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(conf)
+ .compress(reader.getCompression())
+ .version(reader.getFileVersion())
+ .rowIndexStride(reader.getRowIndexStride())
+ .inspector(reader.getObjectInspector());
+ // compression buffer size should only be set if compression is enabled
+ if (reader.getCompression() != org.apache.hadoop.hive.ql.io.orc.CompressionKind.NONE) {
+ writerOptions.bufferSize(reader.getCompressionSize()).enforceBufferSize();
+ }
+
+ try (Writer writer = OrcFile.createWriter(recoveredPath, writerOptions)) {
+
+ // For HIVE-18817, the only thing missing is the last stripe index information.
+ // Get the information from the last stripe and append it to the existing index.
+ // The actual stripe data can be written as-is, similar to OrcFileMergeOperator.
+
+ String keyIndexString = getKeyIndexAsString(reader);
+ if (keyIndexString == null || keyIndexString.equals("null")) {
+ // Key index can be null/"null" if there is only a single stripe. Just start fresh.
+ keyIndexString = "";
+ }
+
+ List<StripeInformation> stripes = reader.getStripes();
+ List<StripeStatistics> stripeStats = reader.getOrcProtoStripeStatistics();
+
+ try (FSDataInputStream inputStream = fs.open(inputPath)) {
+ for (int idx = 0; idx < stripes.size(); ++idx) {
+ // initialize buffer to read the entire stripe.
+ StripeInformation stripe = stripes.get(idx);
+ int stripeLength = (int) stripe.getLength();
+ byte[] buffer = new byte[stripeLength];
+ inputStream.readFully(stripe.getOffset(), buffer, 0, stripeLength);
+
+ // append the stripe buffer to the new ORC file
+ writer.appendStripe(buffer, 0, buffer.length, stripe, stripeStats.get(idx));
+ }
+ }
+
+ // For last stripe we need to get the last trasactionId/bucket/rowId from the last row.
+ long lastRow = reader.getNumberOfRows() - 1;
+ //RecordReader rr = reader.rows();
+ try (RecordReader rr = reader.rows()) {
+ rr.seekToRow(lastRow);
+ OrcStruct row = (OrcStruct) rr.next(null);
+ StructObjectInspector soi = (StructObjectInspector) reader.getObjectInspector();
+ // struct<operation:int,originalTransaction:bigint,bucket:int,rowId:bigint,currentTransaction:bigint
+ List<? extends StructField> structFields = soi.getAllStructFieldRefs();
+
+ StructField transactionField = structFields.get(1);
+ StructField bucketField = structFields.get(2);
+ StructField rowIdField = structFields.get(3);
+
+ long lastTransaction = ((LongObjectInspector) transactionField.getFieldObjectInspector()).get(
+ soi.getStructFieldData(row, transactionField));
+ int lastBucket = ((IntObjectInspector) bucketField.getFieldObjectInspector()).get(
+ soi.getStructFieldData(row, bucketField));
+ long lastRowId = ((LongObjectInspector) rowIdField.getFieldObjectInspector()).get(
+ soi.getStructFieldData(row, rowIdField));
+ keyIndexString += lastTransaction + "," + lastBucket + "," + lastRowId + ";";
+ }
+
+ // Add the rest of the metadata keys.
+ for (String metadataKey : reader.getMetadataKeys()) {
+ if (!metadataKey.equals(OrcRecordUpdater.ACID_KEY_INDEX_NAME)) {
+ writer.addUserMetadata(metadataKey, reader.getMetadataValue(metadataKey));
+ }
+ }
+
+ // Finally add the fixed acid key index.
+ writer.addUserMetadata(OrcRecordUpdater.ACID_KEY_INDEX_NAME, UTF8.encode(keyIndexString));
+ }
+
+ // Confirm the file is really fixed, and replace the old file.
+ Reader newReader = OrcFile.createReader(fs, recoveredPath);
+ boolean fileFixed = isAcidKeyIndexValid(newReader);
+ if (fileFixed) {
+ Path backupDataPath;
+ String scheme = inputPath.toUri().getScheme();
+ String authority = inputPath.toUri().getAuthority();
+ String filePath = inputPath.toUri().getPath();
+
+ // use the same filesystem as input file if backup-path is not explicitly specified
+ if (backup.equals(DEFAULT_BACKUP_PATH)) {
+ backupDataPath = new Path(scheme, authority, DEFAULT_BACKUP_PATH + filePath);
+ } else {
+ backupDataPath = Path.mergePaths(new Path(backup), inputPath);
+ }
+
+ // Move data file to backup path
+ moveFiles(fs, inputPath, backupDataPath);
+ // finally move recovered file to actual file
+ moveFiles(fs, recoveredPath, inputPath);
+
+ System.out.println("Fixed acid key index for " + inputPath);
+ } else {
+ System.out.println("Unable to fix acid key index for " + inputPath);
+ }
+ }
+
+ private static void moveFiles(final FileSystem fs, final Path src, final Path dest)
+ throws IOException {
+ try {
+ // create the dest directory if not exist
+ if (!fs.exists(dest.getParent())) {
+ fs.mkdirs(dest.getParent());
+ }
+
+ // if the destination file exists for some reason delete it
+ fs.delete(dest, false);
+
+ if (fs.rename(src, dest)) {
+ System.err.println("Moved " + src + " to " + dest);
+ } else {
+ throw new IOException("Unable to move " + src + " to " + dest);
+ }
+
+ } catch (Exception e) {
+ throw new IOException("Unable to move " + src + " to " + dest, e);
+ }
+ }
+
+ static String getKeyIndexAsString(Reader reader) {
+ try {
+ ByteBuffer val =
+ reader.getMetadataValue(OrcRecordUpdater.ACID_KEY_INDEX_NAME)
+ .duplicate();
+ return utf8Decoder.decode(val).toString();
+ } catch (CharacterCodingException e) {
+ throw new IllegalArgumentException("Bad string encoding for " +
+ OrcRecordUpdater.ACID_KEY_INDEX_NAME, e);
+ }
+ }
+
+ static Path getRecoveryFile(final Path corruptPath) {
+ return new Path(corruptPath.getParent(), corruptPath.getName() + ".fixacidindex");
+ }
+
+ static Options createOptions() {
+ Options result = new Options();
+
+ result.addOption(OptionBuilder
+ .withLongOpt("check-only")
+ .withDescription("Check acid orc file for valid acid key index and exit without fixing")
+ .create('c'));
+
+ result.addOption(OptionBuilder
+ .withLongOpt("recover")
+ .withDescription("Fix the acid key index for acid orc file if it requires fixing")
+ .create('r'));
+
+ result.addOption(OptionBuilder
+ .withLongOpt("backup-path")
+ .withDescription("specify a backup path to store the corrupted files (default: /tmp)")
+ .hasArg()
+ .create());
+
+ result.addOption(OptionBuilder
+ .withLongOpt("help")
+ .withDescription("print help message")
+ .create('h'));
+
+ return result;
+ }
+
+ public static Collection<String> getAllFilesInPath(final Path path,
+ final Configuration conf) throws IOException {
+ List<String> filesInPath = new ArrayList<>();
+ FileSystem fs = path.getFileSystem(conf);
+ FileStatus fileStatus = fs.getFileStatus(path);
+ if (fileStatus.isDir()) {
+ FileStatus[] fileStatuses = fs.listStatus(path, FileDump.HIDDEN_AND_SIDE_FILE_FILTER);
+ for (FileStatus fileInPath : fileStatuses) {
+ if (fileInPath.isDir()) {
+ filesInPath.addAll(getAllFilesInPath(fileInPath.getPath(), conf));
+ } else {
+ filesInPath.add(fileInPath.getPath().toString());
+ }
+ }
+ } else {
+ filesInPath.add(path.toString());
+ }
+
+ return filesInPath;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hive/blob/d717d385/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestFixAcidKeyIndex.java
----------------------------------------------------------------------
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestFixAcidKeyIndex.java b/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestFixAcidKeyIndex.java
new file mode 100644
index 0000000..bf478cc
--- /dev/null
+++ b/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestFixAcidKeyIndex.java
@@ -0,0 +1,301 @@
+/*
+ * 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.hadoop.hive.ql.io.orc;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetDecoder;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.orc.OrcFile.WriterContext;
+import org.apache.orc.impl.AcidStats;
+import org.apache.orc.impl.OrcAcidUtils;
+import org.apache.orc.impl.WriterImpl;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.Assert.*;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+
+public class TestFixAcidKeyIndex {
+ public final static Logger LOG = LoggerFactory.getLogger(TestFixAcidKeyIndex.class);
+
+ @Rule
+ public TestName testCaseName = new TestName();
+ Path workDir = new Path(System.getProperty("test.tmp.dir","target/tmp"));
+ Configuration conf;
+ Path testFilePath;
+ FileSystem fs;
+
+ @Before
+ public void openFileSystem () throws Exception {
+ conf = new Configuration();
+ fs = FileSystem.getLocal(conf);
+ testFilePath = new Path(workDir, "TestFixAcidKeyIndex." +
+ testCaseName.getMethodName() + ".orc");
+ fs.delete(testFilePath, false);
+ }
+
+ static abstract class TestKeyIndexBuilder
+ extends OrcRecordUpdater.KeyIndexBuilder
+ implements OrcFile.WriterCallback {
+ public TestKeyIndexBuilder(String name) {
+ super(name);
+ }
+
+ // Will be called before closing the ORC file to stop writing any additional information
+ // to the acid key index.
+ abstract void stopWritingKeyIndex();
+ }
+
+ void createTestAcidFile(Path path, int numRows, TestKeyIndexBuilder indexBuilder) throws Exception {
+ FileSystem fs = path.getFileSystem(conf);
+ fs.delete(path, true);
+ String typeStr = "struct<operation:int," +
+ "originalTransaction:bigint,bucket:int,rowId:bigint," +
+ "currentTransaction:bigint," +
+ "row:struct<a:int,b:struct<c:int>,d:string>>";
+ TypeInfo typeInfo = TypeInfoUtils.getTypeInfoFromTypeString(typeStr);
+ Writer writer = OrcFile.createWriter(path,
+ OrcFile.writerOptions(conf)
+ .fileSystem(fs)
+ .inspector(OrcStruct.createObjectInspector(typeInfo))
+ .compress(CompressionKind.NONE)
+ .callback(indexBuilder)
+ .stripeSize(128));
+ // Create ORC file with small stripe size so we can write multiple stripes.
+ OrcStruct row = new OrcStruct(6);
+ row.setFieldValue(0, new IntWritable(0));
+ row.setFieldValue(1, new LongWritable(1));
+ row.setFieldValue(2, new IntWritable(0));
+ LongWritable rowId = new LongWritable();
+ row.setFieldValue(3, rowId);
+ row.setFieldValue(4, new LongWritable(1));
+ OrcStruct rowField = new OrcStruct(3);
+ row.setFieldValue(5, rowField);
+ IntWritable a = new IntWritable();
+ rowField.setFieldValue(0, a);
+ OrcStruct b = new OrcStruct(1);
+ rowField.setFieldValue(1, b);
+ IntWritable c = new IntWritable();
+ b.setFieldValue(0, c);
+ Text d = new Text();
+ rowField.setFieldValue(2, d);
+
+ // Minimum 5000 rows per stripe.
+ for(int r=0; r < numRows; r++) {
+ // row id
+ rowId.set(r);
+ // a
+ a.set(r * 42);
+ // b.c
+ c.set(r * 10001);
+ // d
+ d.set(Integer.toHexString(r));
+ indexBuilder.addKey(OrcRecordUpdater.INSERT_OPERATION, 1, 0, rowId.get());
+ writer.addRow(row);
+ }
+
+ indexBuilder.stopWritingKeyIndex();
+
+ writer.close();
+ }
+
+ void runIndexCheck(Path orcFile, File outFile) throws Exception {
+ // Run with --check-index and save the output to file so it can be checked.
+ PrintStream origOut = System.out;
+ FileOutputStream myOut = new FileOutputStream(outFile);
+
+ System.setOut(new PrintStream(myOut));
+ String[] checkArgs = new String[] {
+ "--check-only",
+ orcFile.toString()
+ };
+ FixAcidKeyIndex.main(checkArgs);
+ System.out.flush();
+ System.setOut(origOut);
+ }
+
+ void checkValidKeyIndex(Path orcFile) throws Exception {
+ String outputFilename = "fixAcidKeyIndex.out";
+ File outFile = new File(workDir.toString(), outputFilename);
+ runIndexCheck(orcFile, outFile);
+
+ // Check the output of FixAcidKeyIndex - it should indicate the index was valid.
+ String outputAsString = FileUtils.readFileToString(outFile);
+ System.out.println(outputAsString);
+ assertTrue(outputAsString.contains("acid key index is valid"));
+ }
+
+ void checkInvalidKeyIndex(Path orcFile) throws Exception {
+ String outputFilename = "fixAcidKeyIndex.out";
+ File outFile = new File(workDir.toString(), outputFilename);
+ runIndexCheck(orcFile, outFile);
+
+ // Check the output of FixAcidKeyIndex - it should indicate the index was invalid.
+ String outputAsString = FileUtils.readFileToString(outFile);
+ System.out.println(outputAsString);
+ assertTrue(outputAsString.contains("acid key index is invalid"));
+ }
+
+ void runFixIndex(Path orcFile, File outFile) throws Exception {
+ // Run with --recover and save the output to a file so it can be checked.
+ PrintStream origOut = System.out;
+ FileOutputStream myOut = new FileOutputStream(outFile);
+
+ System.setOut(new PrintStream(myOut));
+ String[] checkArgs = new String[] {
+ "--recover",
+ orcFile.toString()
+ };
+ FixAcidKeyIndex.main(checkArgs);
+ System.out.flush();
+ System.setOut(origOut);
+ }
+
+ void fixInvalidIndex(Path orcFile) throws Exception {
+ String outputFilename = "fixAcidKeyIndex.out";
+ File outFile = new File(workDir.toString(), outputFilename);
+ runFixIndex(orcFile, outFile);
+
+ // Check the output of FixAcidKeyIndex - it should indicate the index was fixed.
+ String outputAsString = FileUtils.readFileToString(outFile);
+ System.out.println(outputAsString);
+ assertTrue(outputAsString.contains("Fixed acid key index"));
+ }
+
+ void fixValidIndex(Path orcFile) throws Exception {
+ String outputFilename = "fixAcidKeyIndex.out";
+ File outFile = new File(workDir.toString(), outputFilename);
+ runFixIndex(orcFile, outFile);
+
+ // Check the output of FixAcidKeyIndex - it should indicate nothing required fixing.
+ String outputAsString = FileUtils.readFileToString(outFile);
+ System.out.println(outputAsString);
+ assertTrue(outputAsString.contains("No need to recover"));
+ }
+
+ @Test
+ public void testValidKeyIndex() throws Exception {
+ // Try single stripe
+ createTestAcidFile(testFilePath, 100, new GoodKeyIndexBuilder());
+ checkValidKeyIndex(testFilePath);
+ // Attempting to fix a valid - should not result in a new file.
+ fixValidIndex(testFilePath);
+
+ // Multiple stripes
+ createTestAcidFile(testFilePath, 12000, new GoodKeyIndexBuilder());
+ checkValidKeyIndex(testFilePath);
+ // Attempting to fix a valid - should not result in a new file.
+ fixValidIndex(testFilePath);
+ }
+
+ @Test
+ public void testInvalidKeyIndex() throws Exception {
+ // Try single stripe
+ createTestAcidFile(testFilePath, 100, new BadKeyIndexBuilder());
+ checkInvalidKeyIndex(testFilePath);
+ // Try fixing, this should result in new fixed file.
+ fixInvalidIndex(testFilePath);
+
+ // Multiple stripes
+ createTestAcidFile(testFilePath, 12000, new BadKeyIndexBuilder());
+ checkInvalidKeyIndex(testFilePath);
+ // Try fixing, this should result in new fixed file.
+ fixInvalidIndex(testFilePath);
+ }
+
+ @Test
+ public void testNonAcidOrcFile() throws Exception {
+ // Copy data/files/alltypesorc to workDir
+ Path baseSrcDir = new Path(System.getProperty("basedir")).getParent();
+ Path dataFilesPath = new Path(new Path(baseSrcDir, "data"), "files");
+ File origOrcFile = new File(dataFilesPath.toString(), "alltypesorc");
+ File testOrcFile = new File(workDir.toString(), "alltypesorc");
+ FileUtils.copyFile(origOrcFile, testOrcFile);
+
+ String outputFilename = "fixAcidKeyIndex.out";
+ File outFile = new File(workDir.toString(), outputFilename);
+ runIndexCheck(new Path(testOrcFile.getPath()), outFile);
+ String outputAsString = FileUtils.readFileToString(outFile);
+ System.out.println(outputAsString);
+ assertTrue(outputAsString.contains("is not an acid file"));
+ }
+
+ /**
+ * Version of KeyIndexBuilder that generates a valid key index
+ */
+ static class GoodKeyIndexBuilder extends TestKeyIndexBuilder {
+
+ GoodKeyIndexBuilder() {
+ super("GoodKeyIndexBuilder");
+ }
+
+ @Override
+ public void stopWritingKeyIndex() {
+ // Do nothing - this should generate proper index.
+ }
+ }
+
+ /**
+ * Bad version of KeyIndexBuilder which builds an invalid acid key index
+ * by not including the key index info once stopWritingKeyIndex() is called.
+ */
+ static class BadKeyIndexBuilder extends TestKeyIndexBuilder {
+
+ boolean writeAcidIndexInfo = true;
+
+ BadKeyIndexBuilder() {
+ super("BadKeyIndexBuilder");
+ }
+
+ public void stopWritingKeyIndex() {
+ LOG.info("*** Stop writing index!");
+ writeAcidIndexInfo = false;
+ }
+
+ @Override
+ public void preStripeWrite(OrcFile.WriterContext context) throws IOException {
+ LOG.info("*** writeAcidIndexInfo: " + writeAcidIndexInfo);
+ if (!writeAcidIndexInfo) {
+ return;
+ }
+
+ super.preStripeWrite(context);
+ }
+ }
+}