Feature: first version

.gitignore

@@ -0,0 +1,2 @@
+.idea
+target

pom.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>com.aliyun.odps.myJava</groupId>
+    <artifactId>fifo-statement-join</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.aliyun.odps</groupId>
+            <artifactId>odps-sdk-udf</artifactId>
+            <version>0.29.10-public</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+</project>

src/main/java/com/mokamrp/data/udj/FifoStatementJoin.java

@@ -0,0 +1,125 @@
+package com.mokamrp.data.udj;
+
+import com.aliyun.odps.Column;
+import com.aliyun.odps.OdpsType;
+import com.aliyun.odps.Yieldable;
+import com.aliyun.odps.data.ArrayRecord;
+import com.aliyun.odps.data.Record;
+import com.aliyun.odps.udf.DataAttributes;
+import com.aliyun.odps.udf.ExecutionContext;
+import com.aliyun.odps.udf.UDJ;
+import com.aliyun.odps.udf.annotation.Resolve;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * @author wudi
+ */
+@Resolve("->string,string,bigint,bigint") // ID, 指向ID, 分配金额, 分配后余额
+public class FifoStatementJoin extends UDJ {
+    private final static int NEW_ID = 0;
+    private final static int NEW_AMOUNT = 1;
+
+    private final static int REMAIN_ID = 0;
+
+    private final static int REMAIN_BALANCE = 1;
+
+    private final static int RESULT_ID = 0;
+    private final static int RESULT_TARGET = 1;
+    private final static int RESULT_AMOUNT = 2;
+    private final static int RESULT_BALANCE = 3;
+
+    static class RechargeRecord {
+        String id;
+        Long balance;
+
+        RechargeRecord (String id, Long balance) {
+            this.id = id;
+            this.balance = balance;
+        }
+    }
+
+    private Record outputRecord;
+    @Override
+    public void setup(ExecutionContext executionContext, DataAttributes dataAttributes) {
+        System.out.println("FIFO JOIN Queue Version 20230830");
+        outputRecord = new ArrayRecord(new Column[]{
+                new Column("id", OdpsType.STRING),
+                new Column("target_id", OdpsType.STRING),
+                new Column("distributed", OdpsType.BIGINT),
+                new Column("balance", OdpsType.BIGINT),
+        });
+    }
+
+    @Override
+    public void join(Record key, Iterator<Record> left, Iterator<Record> right, Yieldable<Record> output) {
+        // key为关联条件（与ON出现次序一致），这里不用考虑
+        // left 为账单记录，升序排序
+        if (!right.hasNext()) {
+            // 左表为账单记录
+            return;
+        }
+        Queue<RechargeRecord> recharges = new ConcurrentLinkedQueue<>();
+        // right 为分配总表，倒序取各id的最新值
+        left.forEachRemaining(record -> recharges.add(new RechargeRecord(record.getString(REMAIN_ID), record.getBigint(REMAIN_BALANCE))));
+        // emit
+        while (right.hasNext()) {
+            Record record = right.next();
+            String id = record.getString(NEW_ID);
+//            System.out.printf("FIFO JOIN Queue Size: %d%n ID: %s\n", recharges.size(), id);
+            long amount = record.getBigint(NEW_AMOUNT);
+            outputRecord.setString(RESULT_ID, id);
+            if (amount >= 0) {
+                // 充值直接原样照搬一条
+                recharges.add(new RechargeRecord(id, amount));
+//                System.out.println("FIFO JOIN Queue Add: " + id);
+                outputRecord.setString(RESULT_TARGET, id);
+                outputRecord.setBigint(RESULT_AMOUNT, amount);
+                outputRecord.setBigint(RESULT_BALANCE, amount);
+                output.yield(outputRecord);
+            } else {
+                // 消费/退款
+                boolean fail = true;
+                do {
+                    // 一笔消费对应1-N条充值
+                    RechargeRecord current = recharges.peek();
+                    if (current == null) {
+                        System.err.println("FIFO JOIN Queue Empty: " + id);
+                        break;
+                    } else {
+                        fail = false;
+                        outputRecord.setString(RESULT_TARGET, current.id);
+                        if (current.balance >= Math.abs(amount)) {
+                            // 该笔充值够用
+                            current.balance += amount; // amount is negative
+                            outputRecord.setBigint(RESULT_AMOUNT, amount);
+                            outputRecord.setBigint(RESULT_BALANCE, current.balance);
+                            amount = 0;
+                        } else {
+                            // 该笔充值不够用，继续
+                            amount += current.balance;
+                            outputRecord.setBigint(RESULT_AMOUNT, -current.balance);
+                            outputRecord.setBigint(RESULT_BALANCE, 0L);
+                            recharges.remove();
+                        }
+                        output.yield(outputRecord);
+                    }
+                } while (amount != 0);
+                if (fail) {
+                    // 找不到关联，输出空值
+                    outputRecord.set(RESULT_TARGET, null);
+                    outputRecord.set(RESULT_AMOUNT, null);
+                    outputRecord.set(RESULT_BALANCE, null);
+                    output.yield(outputRecord);
+                }
+            }
+        }
+    }
+
+    @Override
+    public void close() {
+
+    }
+}