TheNamesRai commented on code in PR #1813:
URL: https://github.com/apache/phoenix/pull/1813#discussion_r1479237528
##########
phoenix-core/src/main/java/org/apache/phoenix/coprocessor/CDCGlobalIndexRegionScanner.java:
##########
@@ -223,4 +237,85 @@ protected boolean getNextCoveredIndexRow(List<Cell>
result) throws IOException {
}
return false;
}
+
+ private Result getCDCImage(
+ Map<String, Map<String, Object>> preImageObj,
+ Map<String, Map<String, Object>> changeImageObj,
+ boolean isIndexCellDeleteRow, Long indexCellTS, Cell firstCell) {
+ Map<String, Object> rowValueMap = new HashMap<>();
+
+ if (this.cdcChangeScopeSet.size() == 0
+ ||
(this.cdcChangeScopeSet.contains(PTable.CDCChangeScope.PRE))) {
+ rowValueMap.put(PRE_IMAGE, preImageObj);
+ }
+
+ if (this.cdcChangeScopeSet.size() == 0
+ ||
(this.cdcChangeScopeSet.contains(PTable.CDCChangeScope.CHANGE))) {
+ rowValueMap.put(CHANGE_IMAGE, changeImageObj);
+ }
+
+ Map<String, Map<String, Object>> postImageObj = new HashMap<>();
+ if (this.cdcChangeScopeSet.size() == 0
+ ||
(this.cdcChangeScopeSet.contains(PTable.CDCChangeScope.POST))) {
+ if (!isIndexCellDeleteRow) {
+ for (Map.Entry<String, Map<String, Object>> preImageObjFamily
+ : preImageObj.entrySet()) {
+ String columnFamily = preImageObjFamily.getKey();
+ postImageObj.put(columnFamily, new HashMap<>());
+ for (Map.Entry<String, Object> preImageColQual :
+ preImageObjFamily.getValue().entrySet()) {
+
postImageObj.get(columnFamily).put(preImageColQual.getKey(),
+ preImageColQual.getValue());
+ }
+ }
+ for (Map.Entry<String, Map<String, Object>>
changeImageObjFamily
+ : changeImageObj.entrySet()) {
+ String columnFamily = changeImageObjFamily.getKey();
+ if (!postImageObj.containsKey(columnFamily)) {
+ postImageObj.put(columnFamily, new HashMap<>());
+ }
+ for (Map.Entry<String, Object> changeImageColQual :
+ changeImageObjFamily.getValue().entrySet()) {
+
postImageObj.get(columnFamily).put(changeImageColQual.getKey(),
+ changeImageColQual.getValue());
+ }
+ }
+ }
+ rowValueMap.put(POST_IMAGE, postImageObj);
+ }
+
+ if (isIndexCellDeleteRow) {
+ rowValueMap.put(EVENT_TYPE, DELETE_EVENT_TYPE);
+ } else {
+ rowValueMap.put(EVENT_TYPE, UPSERT_EVENT_TYPE);
+ }
+ Gson gson = new GsonBuilder().serializeNulls().create();
+
+ byte[] value =
+ gson.toJson(rowValueMap).getBytes(StandardCharsets.UTF_8);
+ CellBuilder builder =
CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY);
+ Result cdcRow = Result.create(Arrays.asList(builder.
+ setRow(indexToDataRowKeyMap.get(new
ImmutableBytesPtr(firstCell.getRowArray(),
+ firstCell.getRowOffset(), firstCell.getRowLength())
+ .copyBytesIfNecessary())).
+ setFamily(firstCell.getFamilyArray()).
+ setQualifier(scan.getAttribute(CDC_JSON_COL_QUALIFIER)).
+ setTimestamp(indexCellTS).
+ setValue(value).
+ setType(Cell.Type.Put).
+ build()));
+
+ return cdcRow;
+ }
+
+ private Object getColumnValue (Cell cell, PDataType dataType) {
+ if (dataType.getSqlType() == Types.BINARY) {
+ return Base64.getEncoder().encodeToString(cell.getValueArray());
+ } else if (dataType.getSqlType() == Types.DATE) {
+ return ((Date) dataType.toObject(cell.getValueArray())).getTime();
Review Comment:
Change this to Stringified value of Date/Timestamp
--
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]
