iemejia commented on a change in pull request #10815: [BEAM-9279] Make
HBase.ReadAll based on Reads instead of HBaseQuery
URL: https://github.com/apache/beam/pull/10815#discussion_r394877026
##########
File path:
sdks/java/io/hbase/src/main/java/org/apache/beam/sdk/io/hbase/HBaseIO.java
##########
@@ -240,63 +245,109 @@ private Read(
@Override
public void populateDisplayData(DisplayData.Builder builder) {
super.populateDisplayData(builder);
- builder.add(DisplayData.item("configuration",
serializableConfiguration.get().toString()));
+ builder.add(DisplayData.item("configuration", configuration.toString()));
builder.add(DisplayData.item("tableId", tableId));
- builder.addIfNotNull(DisplayData.item("scan",
serializableScan.get().toString()));
+ builder.addIfNotNull(DisplayData.item("scan", scan.toString()));
}
public Configuration getConfiguration() {
- return serializableConfiguration.get();
+ return configuration;
}
public String getTableId() {
return tableId;
}
public Scan getScan() {
- return serializableScan.get();
+ return scan;
}
/** Returns the range of keys that will be read from the table. */
public ByteKeyRange getKeyRange() {
- byte[] startRow = serializableScan.get().getStartRow();
- byte[] stopRow = serializableScan.get().getStopRow();
+ byte[] startRow = scan.getStartRow();
+ byte[] stopRow = scan.getStopRow();
return ByteKeyRange.of(ByteKey.copyFrom(startRow),
ByteKey.copyFrom(stopRow));
}
- private final SerializableConfiguration serializableConfiguration;
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Read read = (Read) o;
+ return configuration.toString().equals(read.configuration.toString())
+ && Objects.equals(tableId, read.tableId)
+ && scan.toString().equals(read.scan.toString());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(configuration, tableId, scan);
+ }
+
+ private Object writeReplace() {
+ return new SerializationProxy(this);
+ }
+
+ private static class SerializationProxy implements Serializable {
+ public SerializationProxy() {}
+
+ public SerializationProxy(Read read) {
+ configuration = read.configuration;
+ tableId = read.tableId;
+ scan = read.scan;
+ }
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ SerializableCoder.of(SerializableConfiguration.class)
+ .encode(new SerializableConfiguration(this.configuration), out);
+ StringUtf8Coder.of().encode(this.tableId, out);
+ ProtobufUtil.toScan(this.scan).writeDelimitedTo(out);
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException {
+ this.configuration =
SerializableCoder.of(SerializableConfiguration.class).decode(in).get();
+ this.tableId = StringUtf8Coder.of().decode(in);
+ this.scan =
ProtobufUtil.toScan(ClientProtos.Scan.parseDelimitedFrom(in));
+ }
+
+ Object readResolve() {
+ return
HBaseIO.read().withConfiguration(configuration).withTableId(tableId).withScan(scan);
+ }
+
+ private Configuration configuration;
+ private String tableId;
+ private Scan scan;
+ }
+
+ @SuppressFBWarnings("SE_BAD_FIELD")
+ private final Configuration configuration;
+
private final String tableId;
- private final SerializableScan serializableScan;
+
+ @SuppressFBWarnings("SE_BAD_FIELD")
+ private final Scan scan;
}
/**
* A {@link PTransform} that works like {@link #read}, but executes read
operations coming from a
- * {@link PCollection} of {@link HBaseQuery}.
+ * {@link PCollection} of {@link Read}.
*/
public static ReadAll readAll() {
- return new ReadAll(null);
+ return new ReadAll();
}
/** Implementation of {@link #readAll}. */
- public static class ReadAll extends PTransform<PCollection<HBaseQuery>,
PCollection<Result>> {
-
- private ReadAll(SerializableConfiguration serializableConfiguration) {
- this.serializableConfiguration = serializableConfiguration;
- }
-
- /** Reads from the HBase instance indicated by the* given configuration. */
- public ReadAll withConfiguration(Configuration configuration) {
- checkArgument(configuration != null, "configuration can not be null");
- return new ReadAll(new SerializableConfiguration(configuration));
- }
+ public static class ReadAll extends PTransform<PCollection<Read>,
PCollection<Result>> {
Review comment:
I better do this in the CHANGES.md release notes file, so this gets
announced with the release notes. Java will cover making users aware at the
code level :)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
