Merge pull request #7 from immutable/features/DATA-1506

DATA-1506 State Diffs POC

src/main/java/io/blockchainetl/ethereum/EthereumPubSubToBigQueryPipeline.java

@@ -9,6 +9,9 @@
 import io.blockchainetl.ethereum.fns.ConvertTokenTransfersToTableRowsFn;
 import io.blockchainetl.ethereum.fns.ConvertTokensToTableRowsFn;
 import io.blockchainetl.ethereum.fns.ConvertTracesToTableRowsFn;
+import io.blockchainetl.ethereum.fns.ConvertStorageDiffsToTableRowsFn;
+import io.blockchainetl.ethereum.fns.ConvertBalanceDiffsToTableRowsFn;
+import io.blockchainetl.ethereum.fns.ConvertNonceDiffsToTableRowsFn;
 import io.blockchainetl.ethereum.fns.ConvertTransactionsToTableRowsFn;
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.transforms.DoFn;
@@ -40,6 +43,9 @@ static void runEthereumPipeline(PubSubToBigQueryPipelineOptions options) {
         entityConfigs.put("logs", ConvertLogsToTableRowsFn.class);
         entityConfigs.put("token_transfers", ConvertTokenTransfersToTableRowsFn.class);
         entityConfigs.put("traces", ConvertTracesToTableRowsFn.class);
+        entityConfigs.put("storage_diffs", ConvertStorageDiffsToTableRowsFn.class);
+        entityConfigs.put("balance_diffs", ConvertBalanceDiffsToTableRowsFn.class);
+        entityConfigs.put("nonce_diffs", ConvertNonceDiffsToTableRowsFn.class);
         entityConfigs.put("contracts", ConvertContractsToTableRowsFn.class);
         entityConfigs.put("tokens", ConvertTokensToTableRowsFn.class);
         runPipeline(options,chainConfigs, entityConfigs);

src/main/java/io/blockchainetl/ethereum/domain/BalanceDiff.java

@@ -0,0 +1,167 @@
+package io.blockchainetl.ethereum.domain;
+
+import com.google.common.base.Objects;
+import com.google.common.base.MoreObjects;
+import org.apache.avro.reflect.Nullable;
+import org.apache.beam.sdk.coders.AvroCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.annotate.JsonProperty;
+
+import java.math.BigInteger;
+
+@DefaultCoder(AvroCoder.class)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class BalanceDiff {
+
+    @Nullable
+    @JsonProperty("address")
+    private String address;
+
+    @Nullable
+    @JsonProperty("from_value")
+    private String fromValue;
+
+    @Nullable
+    @JsonProperty("to_value")
+    private String toValue;
+
+    @Nullable
+    @JsonProperty("transaction_hash")
+    private String transactionHash;
+
+    @Nullable
+    @JsonProperty("transaction_index")
+    private Long transactionIndex;
+
+    @Nullable
+    @JsonProperty("block_timestamp")
+    private Long blockTimestamp;
+
+    @Nullable
+    @JsonProperty("block_number")
+    private Long blockNumber;
+
+    @Nullable
+    @JsonProperty("block_hash")
+    private String blockHash;
+
+    @Nullable
+    @JsonProperty("chain_id")
+    private Long chainId;
+
+    public BalanceDiff() {}
+
+    public String getAddress() {
+        return address;
+    }
+
+    public void setAddress(String address) {
+        this.address = address;
+    }
+
+    public String getFromValue() {
+        return fromValue;
+    }
+
+    public void setFromValue(String fromValue) {
+        this.fromValue = fromValue;
+    }
+
+    public String getToValue() {
+        return toValue;
+    }
+
+    public void setToValue(String toValue) {
+        this.toValue = toValue;
+    }
+
+    public String getTransactionHash() {
+        return transactionHash;
+    }
+
+    public void setTransactionHash(String transactionHash) {
+        this.transactionHash = transactionHash;
+    }
+
+    public Long getTransactionIndex() {
+        return transactionIndex;
+    }
+
+    public void setTransactionIndex(Long transactionIndex) {
+        this.transactionIndex = transactionIndex;
+    }
+
+    public Long getBlockTimestamp() {
+        return blockTimestamp;
+    }
+
+    public void setBlockTimestamp(Long blockTimestamp) {
+        this.blockTimestamp = blockTimestamp;
+    }
+
+    public Long getBlockNumber() {
+        return blockNumber;
+    }
+
+    public void setBlockNumber(Long blockNumber) {
+        this.blockNumber = blockNumber;
+    }
+
+    public String getBlockHash() {
+        return blockHash;
+    }
+
+    public void setBlockHash(String blockHash) {
+        this.blockHash = blockHash;
+    }
+
+    public Long getChainId() {
+        return chainId;
+    }
+
+    public void setChainId(Long chainId) {
+        this.chainId = chainId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        BalanceDiff that = (BalanceDiff) o;
+        return Objects.equal(address, that.address) &&
+            Objects.equal(fromValue, that.fromValue) &&
+            Objects.equal(toValue, that.toValue) &&
+            Objects.equal(transactionHash, that.transactionHash) &&
+            Objects.equal(transactionIndex, that.transactionIndex) &&
+            Objects.equal(blockTimestamp, that.blockTimestamp) &&
+            Objects.equal(blockNumber, that.blockNumber) &&
+            Objects.equal(blockHash, that.blockHash) &&
+            Objects.equal(chainId, that.chainId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(address, fromValue, toValue, transactionHash, transactionIndex, blockTimestamp,
+            blockNumber, blockHash, chainId);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+            .add("address", address)
+            .add("fromValue", fromValue)
+            .add("toValue", toValue)
+            .add("transactionHash", transactionHash)
+            .add("transactionIndex", transactionIndex)
+            .add("blockTimestamp", blockTimestamp)
+            .add("blockNumber", blockNumber)
+            .add("blockHash", blockHash)
+            .add("chainId", chainId)
+            .toString();
+    }
+}

src/main/java/io/blockchainetl/ethereum/domain/NonceDiff.java

@@ -0,0 +1,167 @@
+package io.blockchainetl.ethereum.domain;
+
+import com.google.common.base.Objects;
+import com.google.common.base.MoreObjects;
+import org.apache.avro.reflect.Nullable;
+import org.apache.beam.sdk.coders.AvroCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.codehaus.jackson.annotate.JsonIgnoreProperties;
+import org.codehaus.jackson.annotate.JsonProperty;
+
+import java.math.BigInteger;
+
+@DefaultCoder(AvroCoder.class)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class NonceDiff {
+
+    @Nullable
+    @JsonProperty("address")
+    private String address;
+
+    @Nullable
+    @JsonProperty("from_value")
+    private String fromValue;
+
+    @Nullable
+    @JsonProperty("to_value")
+    private String toValue;
+
+    @Nullable
+    @JsonProperty("transaction_hash")
+    private String transactionHash;
+
+    @Nullable
+    @JsonProperty("transaction_index")
+    private Long transactionIndex;
+
+    @Nullable
+    @JsonProperty("block_timestamp")
+    private Long blockTimestamp;
+
+    @Nullable
+    @JsonProperty("block_number")
+    private Long blockNumber;
+
+    @Nullable
+    @JsonProperty("block_hash")
+    private String blockHash;
+
+    @Nullable
+    @JsonProperty("chain_id")
+    private Long chainId;
+
+    public NonceDiff() {}
+
+    public String getAddress() {
+        return address;
+    }
+
+    public void setAddress(String address) {
+        this.address = address;
+    }
+
+    public String getFromValue() {
+        return fromValue;
+    }
+
+    public void setFromValue(String fromValue) {
+        this.fromValue = fromValue;
+    }
+
+    public String getToValue() {
+        return toValue;
+    }
+
+    public void setToValue(String toValue) {
+        this.toValue = toValue;
+    }
+
+    public String getTransactionHash() {
+        return transactionHash;
+    }
+
+    public void setTransactionHash(String transactionHash) {
+        this.transactionHash = transactionHash;
+    }
+
+    public Long getTransactionIndex() {
+        return transactionIndex;
+    }
+
+    public void setTransactionIndex(Long transactionIndex) {
+        this.transactionIndex = transactionIndex;
+    }
+
+    public Long getBlockTimestamp() {
+        return blockTimestamp;
+    }
+
+    public void setBlockTimestamp(Long blockTimestamp) {
+        this.blockTimestamp = blockTimestamp;
+    }
+
+    public Long getBlockNumber() {
+        return blockNumber;
+    }
+
+    public void setBlockNumber(Long blockNumber) {
+        this.blockNumber = blockNumber;
+    }
+
+    public String getBlockHash() {
+        return blockHash;
+    }
+
+    public void setBlockHash(String blockHash) {
+        this.blockHash = blockHash;
+    }
+
+    public Long getChainId() {
+        return chainId;
+    }
+
+    public void setChainId(Long chainId) {
+        this.chainId = chainId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        NonceDiff that = (NonceDiff) o;
+        return Objects.equal(address, that.address) &&
+            Objects.equal(fromValue, that.fromValue) &&
+            Objects.equal(toValue, that.toValue) &&
+            Objects.equal(transactionHash, that.transactionHash) &&
+            Objects.equal(transactionIndex, that.transactionIndex) &&
+            Objects.equal(blockTimestamp, that.blockTimestamp) &&
+            Objects.equal(blockNumber, that.blockNumber) &&
+            Objects.equal(blockHash, that.blockHash) &&
+            Objects.equal(chainId, that.chainId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(address, fromValue, toValue, transactionHash, transactionIndex, blockTimestamp,
+            blockNumber, blockHash, chainId);
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+            .add("address", address)
+            .add("fromValue", fromValue)
+            .add("toValue", toValue)
+            .add("transactionHash", transactionHash)
+            .add("transactionIndex", transactionIndex)
+            .add("blockTimestamp", blockTimestamp)
+            .add("blockNumber", blockNumber)
+            .add("blockHash", blockHash)
+            .add("chainId", chainId)
+            .toString();
+    }
+}