This is an automated email from the ASF dual-hosted git repository.
abstractdog pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 0bc624ade30 HIVE-28028: Remove duplicated proto reader/writer classes
introduced in HIVE-19288 (#5079) (Raghav Aggarwal reviewed by Laszlo Bodor,
Ayush Saxena)
0bc624ade30 is described below
commit 0bc624ade3020826b4e7101b7ea4789be7e8585b
Author: Raghav Aggarwal <[email protected]>
AuthorDate: Fri Mar 29 23:25:00 2024 +0530
HIVE-28028: Remove duplicated proto reader/writer classes introduced in
HIVE-19288 (#5079) (Raghav Aggarwal reviewed by Laszlo Bodor, Ayush Saxena)
---
itests/hive-unit/pom.xml | 4 +
pom.xml | 39 +++++
ql/pom.xml | 21 +--
.../logging/proto/DatePartitionedLogger.java | 185 ---------------------
.../history/logging/proto/ProtoMessageReader.java | 69 --------
.../logging/proto/ProtoMessageWritable.java | 102 ------------
.../history/logging/proto/ProtoMessageWriter.java | 71 --------
.../dag/history/logging/proto/package-info.java | 23 ---
8 files changed, 48 insertions(+), 466 deletions(-)
diff --git a/itests/hive-unit/pom.xml b/itests/hive-unit/pom.xml
index a1c9be35bb8..011f817a46c 100644
--- a/itests/hive-unit/pom.xml
+++ b/itests/hive-unit/pom.xml
@@ -116,6 +116,10 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-protobuf-history-plugin</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.tez</groupId>
<artifactId>tez-tests</artifactId>
diff --git a/pom.xml b/pom.xml
index 462fcd864d2..1a48f12bf57 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1343,6 +1343,45 @@
<artifactId>tez-common</artifactId>
<version>${tez.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-protobuf-history-plugin</artifactId>
+ <version>${tez.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-jobclient</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-mapreduce-client-common</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-hdfs</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-yarn-client</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-jdbc</artifactId>
diff --git a/ql/pom.xml b/ql/pom.xml
index 5340341b513..5fc3185411b 100644
--- a/ql/pom.xml
+++ b/ql/pom.xml
@@ -721,16 +721,13 @@
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-client</artifactId>
</exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging</artifactId>
- </exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>org.apache.tez</groupId>
+ <artifactId>tez-protobuf-history-plugin</artifactId>
+ <optional>true</optional>
+ </dependency>
<dependency>
<groupId>org.apache.tez</groupId>
<artifactId>tez-mapreduce</artifactId>
@@ -800,14 +797,6 @@
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-yarn-client</artifactId>
</exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging</artifactId>
- </exclusion>
</exclusions>
</dependency>
<dependency>
diff --git
a/ql/src/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java
b/ql/src/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java
deleted file mode 100644
index 58cec7eacea..00000000000
---
a/ql/src/java/org/apache/tez/dag/history/logging/proto/DatePartitionedLogger.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * 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.tez.dag.history.logging.proto;
-
-import java.io.IOException;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
-import java.time.format.DateTimeFormatter;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.yarn.util.Clock;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.protobuf.MessageLite;
-import com.google.protobuf.Parser;
-
-/**
- * Class to create proto reader and writer for a date partitioned directory
structure.
- *
- * @param <T> The proto message type.
- */
-public class DatePartitionedLogger<T extends MessageLite> {
- private static final Logger LOG =
LoggerFactory.getLogger(DatePartitionedLogger.class);
- // Everyone has permission to write, but with sticky set so that delete is
restricted.
- // This is required, since the path is same for all users and everyone
writes into it.
- private static final FsPermission DIR_PERMISSION =
FsPermission.createImmutable((short)01777);
-
- // Since the directories have broad permissions restrict the file read
access.
- private static final FsPermission FILE_UMASK =
FsPermission.createImmutable((short)0066);
-
- private final Parser<T> parser;
- private final Path basePath;
- private final Configuration conf;
- private final Clock clock;
-
- public DatePartitionedLogger(Parser<T> parser, Path baseDir, Configuration
conf, Clock clock)
- throws IOException {
- this.conf = new Configuration(conf);
- this.clock = clock;
- this.parser = parser;
- createDirIfNotExists(baseDir);
- this.basePath = baseDir.getFileSystem(conf).resolvePath(baseDir);
- FsPermission.setUMask(this.conf, FILE_UMASK);
- }
-
- private void createDirIfNotExists(Path path) throws IOException {
- FileSystem fileSystem = path.getFileSystem(conf);
- try {
- if (!fileSystem.exists(path)) {
- fileSystem.mkdirs(path);
- fileSystem.setPermission(path, DIR_PERMISSION);
- }
- } catch (IOException e) {
- // Ignore this exception, if there is a problem it'll fail when trying
to read or write.
- LOG.warn("Error while trying to set permission: ", e);
- }
- }
-
- /**
- * Creates a writer for the given fileName, with date as today.
- */
- public ProtoMessageWriter<T> getWriter(String fileName) throws IOException {
- Path filePath = getPathForDate(getNow().toLocalDate(), fileName);
- return new ProtoMessageWriter<>(conf, filePath, parser);
- }
-
- /**
- * Creates a reader for the given filePath, no validation is done.
- */
- public ProtoMessageReader<T> getReader(Path filePath) throws IOException {
- return new ProtoMessageReader<>(conf, filePath, parser);
- }
-
- /**
- * Create a path for the given date and fileName. This can be used to create
a reader.
- */
- public Path getPathForDate(LocalDate date, String fileName) throws
IOException {
- Path path = new Path(basePath, getDirForDate(date));
- createDirIfNotExists(path);
- return new Path(path, fileName);
- }
-
- public Path getPathForSubdir(String dirName, String fileName) {
- return new Path(new Path(basePath, dirName), fileName);
- }
-
- /**
- * Extract the date from the directory name, this should be a directory
created by this class.
- */
- public LocalDate getDateFromDir(String dirName) {
- if (!dirName.startsWith("date=")) {
- throw new IllegalArgumentException("Invalid directory: "+ dirName);
- }
- return LocalDate.parse(dirName.substring(5),
DateTimeFormatter.ISO_LOCAL_DATE);
- }
-
- /**
- * Returns the directory name for a given date.
- */
- public String getDirForDate(LocalDate date) {
- return "date=" + DateTimeFormatter.ISO_LOCAL_DATE.format(date);
- }
-
- /**
- * Find next available directory, after the given directory.
- */
- public String getNextDirectory(String currentDir) throws IOException {
- // Fast check, if the next day directory exists return it.
- String nextDate = getDirForDate(getDateFromDir(currentDir).plusDays(1));
- FileSystem fileSystem = basePath.getFileSystem(conf);
- if (fileSystem.exists(new Path(basePath, nextDate))) {
- return nextDate;
- }
- // Have to scan the directory to find min date greater than currentDir.
- String dirName = null;
- for (FileStatus status : fileSystem.listStatus(basePath)) {
- String name = status.getPath().getName();
- // String comparison is good enough, since its of form date=yyyy-MM-dd
- if (name.compareTo(currentDir) > 0 && (dirName == null ||
name.compareTo(dirName) < 0)) {
- dirName = name;
- }
- }
- return dirName;
- }
-
- /**
- * Returns new or changed files in the given directory. The offsets are used
to find
- * changed files.
- */
- public List<FileStatus> scanForChangedFiles(String subDir, Map<String, Long>
currentOffsets)
- throws IOException {
- Path dirPath = new Path(basePath, subDir);
- FileSystem fileSystem = basePath.getFileSystem(conf);
- List<FileStatus> newFiles = new ArrayList<>();
- if (!fileSystem.exists(dirPath)) {
- return newFiles;
- }
- for (FileStatus status : fileSystem.listStatus(dirPath)) {
- String fileName = status.getPath().getName();
- Long offset = currentOffsets.get(fileName);
- // If the offset was never added or offset < fileSize.
- if (offset == null || offset < status.getLen()) {
- newFiles.add(status);
- }
- }
- return newFiles;
- }
-
- /**
- * Returns the current time, using the underlying clock in UTC time.
- */
- public LocalDateTime getNow() {
- // Use UTC date to ensure reader date is same on all timezones.
- return LocalDateTime.ofEpochSecond(clock.getTime() / 1000, 0,
ZoneOffset.UTC);
- }
-
- public Configuration getConfig() {
- return conf;
- }
-}
diff --git
a/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageReader.java
b/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageReader.java
deleted file mode 100644
index b56f06673e3..00000000000
---
a/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageReader.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * 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.tez.dag.history.logging.proto;
-
-import java.io.Closeable;
-import java.io.IOException;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.SequenceFile.Reader;
-
-import com.google.protobuf.MessageLite;
-import com.google.protobuf.Parser;
-
-public class ProtoMessageReader<T extends MessageLite> implements Closeable {
- private final Path filePath;
- private final Reader reader;
- private final ProtoMessageWritable<T> writable;
-
- ProtoMessageReader(Configuration conf, Path filePath, Parser<T> parser)
throws IOException {
- this.filePath = filePath;
- // The writer does not flush the length during hflush. Using length
options lets us read
- // past length in the FileStatus but it will throw EOFException during a
read instead
- // of returning null.
- this.reader = new Reader(conf, Reader.file(filePath),
Reader.length(Long.MAX_VALUE));
- this.writable = new ProtoMessageWritable<>(parser);
- }
-
- public Path getFilePath() {
- return filePath;
- }
-
- public void setOffset(long offset) throws IOException {
- reader.seek(offset);
- }
-
- public long getOffset() throws IOException {
- return reader.getPosition();
- }
-
- public T readEvent() throws IOException {
- if (!reader.next(NullWritable.get(), writable)) {
- return null;
- }
- return writable.getMessage();
- }
-
- @Override
- public void close() throws IOException {
- reader.close();
- }
-}
diff --git
a/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWritable.java
b/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWritable.java
deleted file mode 100644
index fe7cb2e0a09..00000000000
---
a/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWritable.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * 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.tez.dag.history.logging.proto;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-
-import com.google.protobuf.ExtensionRegistryLite;
-import org.apache.hadoop.io.Writable;
-
-import com.google.protobuf.CodedInputStream;
-import com.google.protobuf.CodedOutputStream;
-import com.google.protobuf.MessageLite;
-import com.google.protobuf.Parser;
-
-public class ProtoMessageWritable<T extends MessageLite> implements Writable {
- private T message;
- private final Parser<T> parser;
- private DataOutputStream dos;
- private CodedOutputStream cos;
- private DataInputStream din;
- private CodedInputStream cin;
-
- ProtoMessageWritable(Parser<T> parser) {
- this.parser = parser;
- }
-
- public T getMessage() {
- return message;
- }
-
- public void setMessage(T message) {
- this.message = message;
- }
-
- private static class DataOutputStream extends OutputStream {
- DataOutput out;
- @Override
- public void write(int b) throws IOException {
- out.write(b);
- }
-
- @Override
- public void write(byte b[], int off, int len) throws IOException {
- out.write(b, off, len);
- }
- }
-
- @Override
- public void write(DataOutput out) throws IOException {
- if (dos == null) {
- dos = new DataOutputStream();
- cos = CodedOutputStream.newInstance(dos);
- }
- dos.out = out;
- cos.writeMessageNoTag(message);
- cos.flush();
- }
-
- private static class DataInputStream extends InputStream {
- DataInput in;
- @Override
- public int read() throws IOException {
- try {
- return in.readUnsignedByte();
- } catch (EOFException e) {
- return -1;
- }
- }
- }
-
- @Override
- public void readFields(DataInput in) throws IOException {
- if (din == null) {
- din = new DataInputStream();
- cin = CodedInputStream.newInstance(din);
- cin.setSizeLimit(Integer.MAX_VALUE);
- }
- din.in = in;
- message = cin.readMessage(parser, ExtensionRegistryLite.newInstance());
- }
-}
diff --git
a/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWriter.java
b/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWriter.java
deleted file mode 100644
index 9c086ef0d71..00000000000
---
a/ql/src/java/org/apache/tez/dag/history/logging/proto/ProtoMessageWriter.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * 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.tez.dag.history.logging.proto;
-
-import java.io.Closeable;
-import java.io.IOException;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.NullWritable;
-import org.apache.hadoop.io.SequenceFile;
-import org.apache.hadoop.io.SequenceFile.CompressionType;
-import org.apache.hadoop.io.SequenceFile.Writer;
-
-import com.google.protobuf.MessageLite;
-import com.google.protobuf.Parser;
-
-public class ProtoMessageWriter<T extends MessageLite> implements Closeable {
- private final Path filePath;
- private final Writer writer;
- private final ProtoMessageWritable<T> writable;
-
- ProtoMessageWriter(Configuration conf, Path filePath, Parser<T> parser)
throws IOException {
- this.filePath = filePath;
- this.writer = SequenceFile.createWriter(
- conf,
- Writer.file(filePath),
- Writer.keyClass(NullWritable.class),
- Writer.valueClass(ProtoMessageWritable.class),
- Writer.compression(CompressionType.RECORD));
- this.writable = new ProtoMessageWritable<>(parser);
- }
-
- public Path getPath() {
- return filePath;
- }
-
- public long getOffset() throws IOException {
- return writer.getLength();
- }
-
- public void writeProto(T message) throws IOException {
- writable.setMessage(message);
- writer.append(NullWritable.get(), writable);
- }
-
- public void hflush() throws IOException {
- writer.hflush();
- }
-
- @Override
- public void close() throws IOException {
- writer.close();
- }
-}
diff --git
a/ql/src/java/org/apache/tez/dag/history/logging/proto/package-info.java
b/ql/src/java/org/apache/tez/dag/history/logging/proto/package-info.java
deleted file mode 100644
index 23ed46062e0..00000000000
--- a/ql/src/java/org/apache/tez/dag/history/logging/proto/package-info.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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.
- */
-
-/**
- * Logger code copied from tez codebase, this should be removed when we swtich
- * to 0.9.2 tez version and we should depend on the tez libraries for this.
- */
-package org.apache.tez.dag.history.logging.proto;
