Skip to content

Commit

Permalink
HIVE-24958: Create new iceberg-catalog module (Marton Bod reviewed by…
Browse files Browse the repository at this point in the history 
… Laszlo Pinter and Peter Vary)
  • Loading branch information
@marton-bod
marton-bod committed Apr 13, 2021
1 parent e32da3a commit ae3be9a
Show file tree
Hide file tree
Showing 31 changed files with 4,854 additions and 25 deletions.
34 changes: 34 additions & 0 deletions iceberg-catalog/checkstyle/checkstyle-suppressions.xml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
<?xml version="1.0"?>
<!DOCTYPE suppressions PUBLIC
"-//Checkstyle//DTD SuppressionFilter Configuration 1.2//EN"
"https://checkstyle.org/dtds/suppressions_1_2.dtd">

<!-- IMPORTANT ECLIPSE NOTE: If you change this file, you must restart Eclipse
for your changes to take effect in its Checkstyle integration. -->
<suppressions>
<!-- Suppress test classes -->
<suppress files="[/\\]src[/\\].*[Tt]est[/\\](java|groovy)[/\\]" checks="Javadoc*" />
<suppress files="[/\\]src[/\\].*[Tt]est[/\\](java|groovy)[/\\]" checks="VariableDeclarationUsageDistance" />
<suppress files="[/\\]src[/\\].*[Tt]est[/\\](java|groovy)[/\\]" checks="VisibilityModifier" />
<suppress files="[/\\]src[/\\].*[Tt]est[/\\](java|groovy)[/\\]" checks="AvoidStaticImport" />
<suppress files="[/\\]src[/\\].*[Tt]est[/\\](java|groovy)[/\\]" id="BanSystemOut" />
<suppress files="[/\\]src[/\\].*[Tt]est[/\\](java|groovy)[/\\]" id="BanSystemErr" />

<!-- Suppress test resources -->
<suppress files="[/\\]src[/\\]test[/\\]resources.*" checks="." />

<!-- JavadocStyle enforces existence of package-info.java package-level Javadoc; we consider this a bug. -->
<suppress files="package-info.java" checks="JavadocStyle" />

<!-- non-code classes -->
<suppress files="\.(bdr|eot|gif|gzip|jar|json|otf|png|svg|ttf|woff|zip)$" checks="FileTabCharacter" />
<suppress files="\.(bdr|eot|gif|gzip|jar|json|otf|png|svg|ttf|woff|zip)$" checks="NewlineAtEndOfFile" />

<!-- Generated code should not be subjected to checkstyle. -->
<suppress files="[/\\].*[/\\]generated.*[/\\]" checks="." />

<suppress files="org.apache.iceberg.hive.ScriptRunner.java" checks="RegexpHeader"/>

<!-- Referencing guava classes should be allowed in classes within bundled-guava module -->
<suppress files="org.apache.iceberg.GuavaClasses" id="BanUnrelocatedGuavaClasses"/>
</suppressions>
455 changes: 455 additions & 0 deletions iceberg-catalog/checkstyle/checkstyle.xml
View file

Large diffs are not rendered by default.

180 changes: 180 additions & 0 deletions iceberg-catalog/pom.xml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,180 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>hive</artifactId>
<groupId>org.apache.hive</groupId>
<version>4.0.0-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>

<artifactId>hive-iceberg-catalog</artifactId>
<name>Hive Iceberg Catalog</name>

<properties>
<hive.path.to.root>..</hive.path.to.root>
</properties>

<dependencies>
<dependency>
<groupId>org.apache.iceberg</groupId>
<artifactId>iceberg-core</artifactId>
<version>${iceberg.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>${iceberg.avro.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-standalone-metastore-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-serde</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>${hadoop.version}</version>
</dependency>

<!-- test dependencies -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-standalone-metastore-server</artifactId>
<classifier>tests</classifier>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<executions>
<execution>
<goals>
<goal>test-jar</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<showWarnings>true</showWarnings>
<compilerArgs>
<arg>-XDcompilePolicy=simple</arg>
<arg>-Xplugin:ErrorProne -Xep:BanSerializableRead:WARN</arg>
</compilerArgs>
<annotationProcessorPaths>
<path>
<groupId>com.google.errorprone</groupId>
<artifactId>error_prone_core</artifactId>
<version>${iceberg.google.errorprone.version}</version>
</path>
</annotationProcessorPaths>
</configuration>
</plugin>

<plugin>
<groupId>com.diffplug.spotless</groupId>
<artifactId>spotless-maven-plugin</artifactId>
<version>${iceberg.spotless.maven.plugin.version}</version>
<configuration>
<java>
<removeUnusedImports />
<importOrder>
<!-- normal imports, static imports -->
<order>,\# </order>
</importOrder>
<trimTrailingWhitespace/>
<endWithNewline/>
<indent>
<spaces>true</spaces>
<spacesPerTab>2</spacesPerTab>
</indent>
</java>
</configuration>
<executions>
<execution>
<id>validate</id>
<phase>validate</phase>
<goals>
<goal>check</goal>
</goals>
</execution>
</executions>
</plugin>

<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<version>${iceberg.checkstyle.plugin.version}</version>
<configuration>
<propertyExpansion>config_loc=${basedir}/checkstyle/</propertyExpansion>
<configLocation>${basedir}/checkstyle/checkstyle.xml</configLocation>
<suppressionsLocation>${basedir}/checkstyle/checkstyle-suppressions.xml</suppressionsLocation>
<includeTestSourceDirectory>true</includeTestSourceDirectory>
<encoding>UTF-8</encoding>
<consoleOutput>true</consoleOutput>
<failsOnError>true</failsOnError>
</configuration>
<executions>
<execution>
<id>validate</id>
<phase>validate</phase>
<goals>
<goal>check</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>

<profiles>
<profile>
<id>jdk8</id>
<activation>
<jdk>1.8</jdk>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<fork>true</fork>
<compilerArgs combine.children="append">
<arg>-J-Xbootclasspath/p:${settings.localRepository}/com/google/errorprone/javac/${iceberg.google.errorprone.javac.version}/javac-${iceberg.google.errorprone.javac.version}.jar</arg>
</compilerArgs>
</configuration>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>
148 changes: 148 additions & 0 deletions iceberg-catalog/src/main/java/org/apache/iceberg/hive/ClientPool.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
/*
* 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.iceberg.hive;

import java.io.Closeable;
import java.util.ArrayDeque;
import java.util.Deque;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class ClientPool<C, E extends Exception> implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(ClientPool.class);

private final int poolSize;
private final Deque<C> clients;
private final Class<? extends E> reconnectExc;
private final Object signal = new Object();
private volatile int currentSize;
private boolean closed;

ClientPool(int poolSize, Class<? extends E> reconnectExc) {
this.poolSize = poolSize;
this.reconnectExc = reconnectExc;
this.clients = new ArrayDeque<>(poolSize);
this.currentSize = 0;
this.closed = false;
}

public interface Action<R, C, E extends Exception> {
R run(C client) throws E;
}

public <R> R run(Action<R, C, E> action) throws E, InterruptedException {
C client = get();
try {
return action.run(client);

} catch (Exception exc) {
if (isConnectionException(exc)) {
try {
client = reconnect(client);
} catch (Exception ignored) {
// if reconnection throws any exception, rethrow the original failure
throw reconnectExc.cast(exc);
}

return action.run(client);
}

throw exc;

} finally {
release(client);
}
}

protected abstract C newClient();

protected abstract C reconnect(C client);

protected boolean isConnectionException(Exception exc) {
return reconnectExc.isInstance(exc);
}

protected abstract void close(C client);

@Override
public void close() {
this.closed = true;
try {
while (currentSize > 0) {
if (!clients.isEmpty()) {
synchronized (this) {
if (!clients.isEmpty()) {
C client = clients.removeFirst();
close(client);
currentSize -= 1;
}
}
}
if (clients.isEmpty() && currentSize > 0) {
// wake every second in case this missed the signal
synchronized (signal) {
signal.wait(1000);
}
}
}

} catch (InterruptedException e) {
Thread.currentThread().interrupt();
LOG.warn("Interrupted while shutting down pool. Some clients may not be closed.", e);
}
}

@VisibleForTesting
int poolSize() {
return poolSize;
}

private C get() throws InterruptedException {
Preconditions.checkState(!closed, "Cannot get a client from a closed pool");
while (true) {
if (!clients.isEmpty() || currentSize < poolSize) {
synchronized (this) {
if (!clients.isEmpty()) {
return clients.removeFirst();
} else if (currentSize < poolSize) {
C client = newClient();
currentSize += 1;
return client;
}
}
}
synchronized (signal) {
// wake every second in case this missed the signal
signal.wait(1000);
}
}
}

private void release(C client) {
synchronized (this) {
clients.addFirst(client);
}
synchronized (signal) {
signal.notify();
}
}
}

0 comments on commit ae3be9a

Please sign in to comment.