jackye1995 commented on a change in pull request #3376: URL: https://github.com/apache/iceberg/pull/3376#discussion_r743059370
########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsClientFactory.java ########## @@ -0,0 +1,95 @@ +/* + * 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.dell; + +import com.emc.object.s3.S3Client; +import com.emc.object.s3.S3Config; +import com.emc.object.s3.jersey.S3JerseyClient; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.net.URI; +import java.util.Map; +import java.util.Optional; + +public interface EcsClientFactory { + + /** + * Create the ECS S3 Client from properties + */ + static S3Client create(Map<String, String> properties) { + return createWithFactory(properties).orElseGet(() -> createDefault(properties)); + } + + /** + * Try to create the ECS S3 client from factory method. + */ + static Optional<S3Client> createWithFactory(Map<String, String> properties) { Review comment: we have util classes like `DynConstructors` for this purpose you can use, which can simplify the code a lot, see https://github.com/apache/iceberg/blob/master/core/src/main/java/org/apache/iceberg/CatalogUtil.java#L166-L195 as an example. ########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsClientFactory.java ########## @@ -0,0 +1,95 @@ +/* + * 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.dell; + +import com.emc.object.s3.S3Client; +import com.emc.object.s3.S3Config; +import com.emc.object.s3.jersey.S3JerseyClient; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.net.URI; +import java.util.Map; +import java.util.Optional; + +public interface EcsClientFactory { + + /** + * Create the ECS S3 Client from properties + */ + static S3Client create(Map<String, String> properties) { + return createWithFactory(properties).orElseGet(() -> createDefault(properties)); + } + + /** + * Try to create the ECS S3 client from factory method. + */ + static Optional<S3Client> createWithFactory(Map<String, String> properties) { + String factory = properties.get(EcsClientProperties.ECS_CLIENT_FACTORY); + if (factory == null || factory.isEmpty()) { + return Optional.empty(); + } + + String[] classAndMethod = factory.split("#", 2); + if (classAndMethod.length != 2) { + throw new IllegalArgumentException(String.format("invalid property %s=%s", + EcsClientProperties.ECS_CLIENT_FACTORY, factory)); + } + + Class<?> clazz; + try { + clazz = Class.forName(classAndMethod[0], true, Thread.currentThread().getContextClassLoader()); + } catch (ClassNotFoundException e) { + throw new IllegalArgumentException( + String.format("invalid property %s=%s", EcsClientProperties.ECS_CLIENT_FACTORY, factory), + e); + } + + S3Client client; + try { + client = (S3Client) MethodHandles.lookup() + .findStatic(clazz, classAndMethod[1], MethodType.methodType(S3Client.class, Map.class)) + .invoke(properties); + } catch (Throwable e) { + throw new IllegalArgumentException( + String.format("invalid property %s=%s that throw exception", EcsClientProperties.ECS_CLIENT_FACTORY, factory), + e); + } + + if (client == null) { + throw new IllegalArgumentException(String.format( + "invalid property %s=%s that return null client", + EcsClientProperties.ECS_CLIENT_FACTORY, factory)); + } + + return Optional.of(client); + } + + /** + * Get built-in ECS S3 client. + */ + static S3Client createDefault(Map<String, String> properties) { + S3Config config = new S3Config(URI.create(properties.get(EcsClientProperties.ENDPOINT))); + + config.withIdentity(properties.get(EcsClientProperties.ACCESS_KEY_ID)) Review comment: I think you have to check these properties are non-null? ########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsFileIO.java ########## @@ -0,0 +1,104 @@ +/* + * 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.dell; + +import com.emc.object.s3.S3Client; +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Map; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link java.io.Externalizable} FileIO of ECS S3 object client. + */ +public class EcsFileIO implements FileIO, Externalizable, AutoCloseable { + + private static final Logger log = LoggerFactory.getLogger(EcsFileIO.class); Review comment: nit: LOG instead of log ########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsFileIO.java ########## @@ -0,0 +1,104 @@ +/* + * 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.dell; + +import com.emc.object.s3.S3Client; +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Map; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link java.io.Externalizable} FileIO of ECS S3 object client. + */ +public class EcsFileIO implements FileIO, Externalizable, AutoCloseable { + + private static final Logger log = LoggerFactory.getLogger(EcsFileIO.class); + + /** + * Saved properties for {@link java.io.Serializable} + */ + private Map<String, String> properties; + private S3Client client; + + /** + * Blank constructor + */ + public EcsFileIO() { + } + + @Override + public void initialize(Map<String, String> inputProperties) { + this.properties = ImmutableMap.copyOf(inputProperties); + this.client = EcsClientFactory.create(inputProperties); + } + + @Override + public InputFile newInputFile(String path) { + return new EcsInputFile(client, path); + } + + @Override + public OutputFile newOutputFile(String path) { + return new EcsOutputFile(client, path); + } + + @Override + public void deleteFile(String path) { + EcsURI uri = EcsURI.create(path); + client.deleteObject(uri.getBucket(), uri.getName()); + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException { + out.writeObject(properties); + } + + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + @SuppressWarnings("unchecked") + Map<String, String> inputProperties = (Map<String, String>) in.readObject(); + initialize(inputProperties); + } + + @Override + public void close() { + client.destroy(); + log.info("FileIO closed"); + } + + @VisibleForTesting + Map<String, String> getProperties() { Review comment: nit: prefers methods to have no `get` ########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsOutputFile.java ########## @@ -0,0 +1,68 @@ +/* + * 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.dell; + +import com.emc.object.s3.S3Client; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.PositionOutputStream; + +public class EcsOutputFile implements OutputFile { + + private final S3Client client; + private final String location; + private final EcsURI uri; + + public EcsOutputFile(S3Client client, String location) { + this.client = client; + this.location = location; + this.uri = EcsURI.create(location); + } + + /** + * Check object existence and then create a {@link PositionOutputStream} + * + * @return Output stream of object + */ + @Override + public PositionOutputStream create() { + if (!toInputFile().exists()) { Review comment: to follow the pattern with other input and output files, have a `BaseEcsFile` with method `exists()`, so `InputFile` gets the method by extending it, and you can directly check exists in output file without converting the file back to an input file. ########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsFileIO.java ########## @@ -0,0 +1,104 @@ +/* + * 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.dell; + +import com.emc.object.s3.S3Client; +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Map; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link java.io.Externalizable} FileIO of ECS S3 object client. + */ +public class EcsFileIO implements FileIO, Externalizable, AutoCloseable { + + private static final Logger log = LoggerFactory.getLogger(EcsFileIO.class); + + /** + * Saved properties for {@link java.io.Serializable} + */ + private Map<String, String> properties; + private S3Client client; + + /** + * Blank constructor + */ + public EcsFileIO() { + } + + @Override + public void initialize(Map<String, String> inputProperties) { + this.properties = ImmutableMap.copyOf(inputProperties); + this.client = EcsClientFactory.create(inputProperties); + } + + @Override + public InputFile newInputFile(String path) { + return new EcsInputFile(client, path); + } + + @Override + public OutputFile newOutputFile(String path) { + return new EcsOutputFile(client, path); + } + + @Override + public void deleteFile(String path) { + EcsURI uri = EcsURI.create(path); + client.deleteObject(uri.getBucket(), uri.getName()); + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException { + out.writeObject(properties); + } + + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + @SuppressWarnings("unchecked") + Map<String, String> inputProperties = (Map<String, String>) in.readObject(); + initialize(inputProperties); + } + + @Override + public void close() { + client.destroy(); + log.info("FileIO closed"); + } + + @VisibleForTesting + Map<String, String> getProperties() { + return properties; + } + + @VisibleForTesting + S3Client getClient() { Review comment: nit: prefers methods to have no get ########## File path: dell/src/test/java/org/apache/iceberg/dell/EcsAppendOutputStreamTest.java ########## @@ -0,0 +1,88 @@ +/* + * 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.dell; + +import com.emc.object.Range; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import org.apache.iceberg.dell.mock.EcsS3MockRule; +import org.apache.iceberg.relocated.com.google.common.io.ByteStreams; +import org.junit.Rule; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class EcsAppendOutputStreamTest { + + @Rule + public EcsS3MockRule rule = EcsS3MockRule.create(); + + @Test + public void generalTest() throws IOException { + String objectName = "test"; + try (EcsAppendOutputStream output = EcsAppendOutputStream.createWithBufferSize( + rule.getClient(), + new EcsURI(rule.getBucket(), objectName), + 10)) { + // write 1 byte + output.write('1'); + // write 3 bytes + output.write("123".getBytes()); + // write 7 bytes, totally 11 bytes > local buffer limit (10) + output.write("1234567".getBytes()); + // write 11 bytes, flush remain 7 bytes and new 11 bytes + output.write("12345678901".getBytes()); + } + + try (InputStream input = rule.getClient().readObjectStream(rule.getBucket(), objectName, + Range.fromOffset(0))) { + assertEquals("object content", "1" + "123" + "1234567" + "12345678901", Review comment: nit: prefers more concrete assert message, like "Must write all the object content" ########## File path: dell/src/test/java/org/apache/iceberg/dell/EcsAppendOutputStreamTest.java ########## @@ -0,0 +1,88 @@ +/* + * 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.dell; + +import com.emc.object.Range; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import org.apache.iceberg.dell.mock.EcsS3MockRule; +import org.apache.iceberg.relocated.com.google.common.io.ByteStreams; +import org.junit.Rule; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class EcsAppendOutputStreamTest { + + @Rule + public EcsS3MockRule rule = EcsS3MockRule.create(); + + @Test + public void generalTest() throws IOException { Review comment: nit: prefers test method names to be `testXXX`, also prefers more specific test names like `testBaiscEcsAppendOutputStreamWrite`. This helps when other people to understand the content of failed tests. Could you change all the names in tests? Thanks!! ########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsFileIO.java ########## @@ -0,0 +1,104 @@ +/* + * 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.dell; + +import com.emc.object.s3.S3Client; +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.Map; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * A {@link java.io.Externalizable} FileIO of ECS S3 object client. + */ +public class EcsFileIO implements FileIO, Externalizable, AutoCloseable { + + private static final Logger log = LoggerFactory.getLogger(EcsFileIO.class); + + /** + * Saved properties for {@link java.io.Serializable} + */ + private Map<String, String> properties; + private S3Client client; + + /** + * Blank constructor + */ + public EcsFileIO() { + } + + @Override + public void initialize(Map<String, String> inputProperties) { + this.properties = ImmutableMap.copyOf(inputProperties); + this.client = EcsClientFactory.create(inputProperties); + } + + @Override + public InputFile newInputFile(String path) { + return new EcsInputFile(client, path); + } + + @Override + public OutputFile newOutputFile(String path) { + return new EcsOutputFile(client, path); + } + + @Override + public void deleteFile(String path) { + EcsURI uri = EcsURI.create(path); + client.deleteObject(uri.getBucket(), uri.getName()); + } + + @Override + public void writeExternal(ObjectOutput out) throws IOException { + out.writeObject(properties); + } + + @Override + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + @SuppressWarnings("unchecked") + Map<String, String> inputProperties = (Map<String, String>) in.readObject(); + initialize(inputProperties); + } + + @Override + public void close() { + client.destroy(); + log.info("FileIO closed"); Review comment: logging message too generic, does not seem necessary. Could you remove it or make it more informational? ########## File path: dell/src/main/java/org/apache/iceberg/dell/EcsSeekableInputStream.java ########## @@ -0,0 +1,110 @@ +/* + * 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.dell; + +import com.emc.object.Range; +import com.emc.object.s3.S3Client; +import java.io.IOException; +import java.io.InputStream; +import org.apache.iceberg.io.SeekableInputStream; + +/** + * A {@link SeekableInputStream} impl that warp {@link S3Client#readObjectStream(String, String, Range)} Review comment: nit: no abbreviation, `implementation` instead of `impl`. ########## File path: dell/src/test/java/org/apache/iceberg/dell/EcsAppendOutputStreamTest.java ########## @@ -0,0 +1,88 @@ +/* + * 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.dell; + +import com.emc.object.Range; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import org.apache.iceberg.dell.mock.EcsS3MockRule; +import org.apache.iceberg.relocated.com.google.common.io.ByteStreams; +import org.junit.Rule; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class EcsAppendOutputStreamTest { Review comment: nit: prefers tests classes named with `TestXXX` (I also need to change it for the AWS module...) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
