Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 24 additions & 2 deletions hadoop-ozone/dist/src/shell/ozone/ozone
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,16 @@ MYNAME="${BASH_SOURCE-$0}"
bin=$(cd -P -- "$(dirname -- "${MYNAME}")" >/dev/null && pwd -P)
JVM_PID="$$"

## @description true when the ozone-iceberg artifact is present in this distribution
## @audience private
function ozone_iceberg_available
{
local lib_dir="${HDDS_LIB_JARS_DIR:-${OZONE_HOME}/share/ozone/lib}"

[[ -f "${OZONE_HOME}/share/ozone/classpath/ozone-iceberg.classpath" ]] \
&& compgen -G "${lib_dir}/ozone-iceberg-*.jar" > /dev/null
}

## @description build up the ozone command's usage text.
## @audience public
## @stability stable
Expand Down Expand Up @@ -65,7 +75,9 @@ function ozone_usage
ozone_add_subcommand "repair" client "Ozone repair tool"
ozone_add_subcommand "ratis" client "Ozone ratis tool"
ozone_add_subcommand "vapor" client "Ozone server simulator"

if ozone_iceberg_available; then
ozone_add_subcommand "iceberg" client "commands for Iceberg tables on Ozone (see ozone iceberg --help for subcommands)"
fi
ozone_generate_usage "${OZONE_SHELL_EXECNAME}" false
}

Expand Down Expand Up @@ -252,6 +264,15 @@ function ozonecmd_case
OZONE_VAPOR_OPTS="${OZONE_VAPOR_OPTS} ${RATIS_OPTS} ${OZONE_MODULE_ACCESS_ARGS}"
OZONE_RUN_ARTIFACT_NAME="ozone-vapor"
;;
iceberg)
if ! ozone_iceberg_available; then
ozone_error "ERROR: ozone iceberg is not available in this distribution (requires JDK 11+ build)."
exit 1
fi
OZONE_CLASSNAME="org.apache.hadoop.ozone.iceberg.IcebergCommand"
OZONE_RUN_ARTIFACT_NAME="ozone-iceberg"
OZONE_SUBCMD_SUPPORTDAEMONIZATION=false
;;
*)
OZONE_CLASSNAME="${subcmd}"
if ! ozone_validate_classname "${OZONE_CLASSNAME}"; then
Expand Down Expand Up @@ -288,7 +309,8 @@ function ozone_suppress_shell_log
{
if [[ "${OZONE_RUN_ARTIFACT_NAME}" =~ ozone-cli-.* ]] \
|| [[ "${OZONE_RUN_ARTIFACT_NAME}" == "ozone-dist" ]] \
|| [[ "${OZONE_RUN_ARTIFACT_NAME}" == "ozone-tools" ]]; then
|| [[ "${OZONE_RUN_ARTIFACT_NAME}" == "ozone-tools" ]] \
|| [[ "${OZONE_RUN_ARTIFACT_NAME}" == "ozone-iceberg" ]]; then
if [[ -z "${OZONE_ORIGINAL_LOGLEVEL}" ]] \
&& [[ -z "${OZONE_ORIGINAL_ROOT_LOGGER}" ]]; then
OZONE_LOGLEVEL=OFF
Expand Down
35 changes: 25 additions & 10 deletions hadoop-ozone/iceberg/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,19 @@
</properties>

<dependencies>
<dependency>
<groupId>info.picocli</groupId>
<artifactId>picocli</artifactId>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>1.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
</dependency>

<!-- Iceberg dependencies -->
<dependency>
Expand Down Expand Up @@ -101,12 +109,20 @@
</exclusion>
</exclusions>
</dependency>

<dependency>
<groupId>org.apache.ozone</groupId>
<artifactId>hdds-cli-common</artifactId>
</dependency>
<dependency>
<groupId>org.apache.ozone</groupId>
<artifactId>hdds-common</artifactId>
</dependency>
<dependency>
<groupId>org.apache.parquet</groupId>
<artifactId>parquet-column</artifactId>
<version>1.16.0</version>
</dependency>

<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
Expand Down Expand Up @@ -175,15 +191,14 @@
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
</exclusion>
</exclusions>
<groupId>org.apache.ozone</groupId>
<artifactId>ozone-filesystem</artifactId>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-reload4j</artifactId>
<scope>runtime</scope>
</dependency>
</dependencies>

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
/*
* 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.ozone.iceberg;

import org.apache.hadoop.hdds.cli.GenericCli;
import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import picocli.CommandLine.Command;

/**
* Parent command for Iceberg tables on Ozone.
*/
@Command(
name = "ozone iceberg",
aliases = "iceberg",
description = "commands for Iceberg tables on Ozone",
subcommands = {
RewriteTablePathCommand.class
},
versionProvider = HddsVersionProvider.class,
mixinStandardHelpOptions = true
)
public class IcebergCommand extends GenericCli {

public static void main(String[] args) {
new IcebergCommand().run(args);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
/*
* 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.ozone.iceberg;

import java.util.concurrent.Callable;
import org.apache.hadoop.hdds.cli.AbstractSubcommand;
import org.apache.iceberg.Table;
import org.apache.iceberg.actions.RewriteTablePath;
import org.apache.iceberg.hadoop.HadoopTables;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;

/**
* CLI to rewrite Iceberg table paths.
*/
@Command(
name = "rewrite-path",
description = "Rewrite Iceberg table paths for table migration"
)
public class RewriteTablePathCommand extends AbstractSubcommand implements Callable<Void> {

@Option(
names = {"-l", "--table-location"},
required = true,
description = "The latest metadata.json file path of the table"
)
private String tableLocation;

@Option(
names = {"-s", "--source-prefix"},
required = true,
description = "Source path prefix to replace"
)
private String sourcePrefix;

@Option(
names = {"-t", "--target-prefix"},
required = true,
description = "Target path prefix"
)
private String targetPrefix;

@Option(
names = {"--staging"},
description = "Staging location where all the rewritten files will be placed "
+ "(Default is a new directory under the table's current metadata directory.)"
)
private String stagingLocation;

@Option(
names = {"--start-version"},
description = "Start version metadata file name (optional, e.g., v1.metadata.json)"
)
private String startVersion;

@Option(
names = {"--end-version"},
description = "End version metadata file name (optional, defaults to current)"
)
private String endVersion;

@Option(
names = {"--threads"},
defaultValue = "10",
description = "Number of threads to use (positive integer). "
+ "If omitted or zero, the default thread count 10 is used."
)
private int threads;

@Override
public Void call() {
out().println("Starting Iceberg table path rewrite");
out().println("Table location: " + tableLocation);
out().println("Source prefix: " + sourcePrefix);
out().println("Target prefix: " + targetPrefix);

HadoopTables tables = new HadoopTables(getOzoneConf());
Table table = tables.load(tableLocation.trim());
out().println("Table loaded: " + table.location());

RewriteTablePathOzoneAction action = new RewriteTablePathOzoneAction(table, threads);
out().println("Threads: " + threads);

RewriteTablePath rewriteAction = action.rewriteLocationPrefix(sourcePrefix, targetPrefix);

if (stagingLocation != null && !stagingLocation.isBlank()) {
out().println("Staging location: " + stagingLocation);
rewriteAction.stagingLocation(stagingLocation);
}

if (startVersion != null && !startVersion.isBlank()) {
out().println("Start version: " + startVersion);
rewriteAction.startVersion(startVersion);
}

if (endVersion != null && !endVersion.isBlank()) {
out().println("End version: " + endVersion);
rewriteAction.endVersion(endVersion);
}

RewriteTablePath.Result result = rewriteAction.execute();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add execute inside try/catch and output if any error to console, like if validateInputs fails.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not addressed yet?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have made IcebergCommand extend GenericCli due to which
rewriteAction.execute() errors are handled by GenericCli's PicoCLI execution exception handler, which prints the error message and also provides --verbose with which we can see the full stack trace. So an explicit try/catch in RewriteTablePathCommand would be redundant.


out().println();
out().println("Rewrite completed successfully");
out().println(" Latest version: " + result.latestVersion());
out().println(" Staging location: " + result.stagingLocation());
out().println();
out().println("Next step: Copy files from source to target using the file list");
out().println(" File list location: " + result.fileListLocation());
return null;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -96,22 +96,16 @@ public class RewriteTablePathOzoneAction implements RewriteTablePath {
private String startVersionName;
private String endVersionName;
private String stagingDir;
private int parallelism;
private int threads;
Comment thread
adoroszlai marked this conversation as resolved.

private ExecutorService executorService;
private static final int MAX_INFLIGHT_MULTIPLIER = 4;
private static final int DEFAULT_THREAD_COUNT = 10;

private final Table table;

public RewriteTablePathOzoneAction(Table table) {
public RewriteTablePathOzoneAction(Table table, int threads) {
this.table = table;
this.parallelism = DEFAULT_THREAD_COUNT;
}

public RewriteTablePathOzoneAction(Table table, int parallelism) {
this.table = table;
this.parallelism = parallelism;
this.threads = threads;
}

@Override
Expand Down Expand Up @@ -147,7 +141,7 @@ public RewriteTablePath stagingLocation(String stagingLocation) {
@Override
public Result execute() {
validateInputs();
executorService = Executors.newFixedThreadPool(parallelism);
executorService = Executors.newFixedThreadPool(threads);
try {
return doExecute();
} finally {
Expand Down Expand Up @@ -326,7 +320,7 @@ private Set<Pair<String, String>> rewriteVersionFile(TableMetadata metadata, Str
private Set<String> manifestsToRewrite(Set<Snapshot> validSnapshots, Set<Long> deltaSnapshotIds) {

Set<String> manifestPaths = ConcurrentHashMap.newKeySet();
int maxInFlight = parallelism * MAX_INFLIGHT_MULTIPLIER;
int maxInFlight = threads * MAX_INFLIGHT_MULTIPLIER;
Semaphore semaphore = new Semaphore(maxInFlight);

ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(executorService);
Expand Down Expand Up @@ -441,7 +435,7 @@ private RewriteResult<ManifestFile> rewriteManifestLists(Set<Snapshot> validSnap
return new RewriteResult<>();
}

int maxInFlight = parallelism * MAX_INFLIGHT_MULTIPLIER;
int maxInFlight = threads * MAX_INFLIGHT_MULTIPLIER;
Semaphore semaphore = new Semaphore(maxInFlight);
ExecutorCompletionService<RewriteResult<ManifestFile>> completionService =
new ExecutorCompletionService<>(executorService);
Expand Down Expand Up @@ -535,7 +529,7 @@ private RewriteContentFileResult rewriteManifests(
return new RewriteContentFileResult();
}

int maxInFlight = parallelism * MAX_INFLIGHT_MULTIPLIER;
int maxInFlight = threads * MAX_INFLIGHT_MULTIPLIER;
Semaphore semaphore = new Semaphore(maxInFlight);
ExecutorCompletionService<RewriteContentFileResult> completionService =
new ExecutorCompletionService<>(executorService);
Expand Down Expand Up @@ -735,7 +729,7 @@ private void rewritePositionDeletes(Set<DeleteFile> toRewrite) {
}

RewriteTablePathUtil.PositionDeleteReaderWriter posDeleteReaderWriter = new OzonePositionDeleteReaderWriter();
int maxInFlight = parallelism * MAX_INFLIGHT_MULTIPLIER;
int maxInFlight = threads * MAX_INFLIGHT_MULTIPLIER;
Semaphore semaphore = new Semaphore(maxInFlight);
ExecutorCompletionService<Void> completionService = new ExecutorCompletionService<>(executorService);
int submittedTasks = 0;
Expand Down
Loading