DEV Community: Apache ShardingSphere

Fuzzy query for CipherColumn | ShardingSphere 5.3.0 Deep Dive

Apache ShardingSphere — Thu, 29 Dec 2022 02:30:01 +0000

1. Background

Apache ShardingSphere supports data encryption. By parsing users’ SQL input and rewriting the SQL according to the users’ encryption rules, the original data is encrypted and stored with ciphertext data in the underlying database at the same time.

When a user queries the data, it only fetches the ciphertext data from the database, decrypts it, and finally returns the decrypted original data to the user. However, because the encryption algorithm encrypts the whole string, fuzzy queries cannot be achieved.

Nevertheless, many businesses still need fuzzy queries after the data is encrypted. In version 5.3.0, Apache ShardingSphere provides users with a default fuzzy query algorithm, supporting the fuzzy query for encrypted fields. The algorithm also supports hot plugging, which can be customized by users, and the fuzzy query can be achieved through configuration.

2. How to achieve fuzzy query in encrypted scenarios?

2.1 Load data to the in-memory database (IMDB)

Load all the data into the IMDB to decrypt it; then it’ll be like querying the original data. This method can achieve fuzzy queries. If the amount of data is small, this method will prove to be simple and cost-effective, while on the other hand, if the amount of data is large, it’ll turn out to be a disaster.

2.2 Implement encryption & decryption functions consistent with database programs

The second method is to modify fuzzy query conditions and use the database decryption function to decrypt data first and then implement fuzzy query. This method’s advantage is the low implementation & development cost, as well as use cost.

Users only need to slightly modify the previous fuzzy query conditions. However, the ciphertext and encryption functions are stored together in the database, which cannot cope with the problem of account data leaks.

Native SQL: select * from user where name like "%xxx%" 
After implementing the decryption function: ѕеlесt * frоm uѕеr whеrе dесоdе(namе) lіkе "%ххх%"

2.3 Store after data masking

Implement data masking on ciphertext and then store it in a fuzzy query column. This method could lack in terms of precision.

For example, mobile number 13012345678 becomes 130****5678 after the masking algorithm is performed.

2.4 Perform encrypted storage after tokenization and combination

This method performs tokenization and combination on ciphertext data and then encrypts the resultset by grouping characters with fixed length and splitting a field into multiple ones. For example, we take four English characters and two Chinese characters as a query condition:

ningyu1 uses the 4-character as a group to encrypt, so the first group is ning, the second group ingy, the third group ngyu, the fourth group gyu1, and so on. All the characters are encrypted and stored in the fuzzy query column. If you want to retrieve all data that contains four characters, such as ingy, encrypt the characters and use a key like"%partial%" to query.

Shortcomings:

Increased storage costs: free grouping will increase the amount of data and the data length will increase after being encrypted.
Limited length in fuzzy query: due to security issues, the length of free grouping cannot be too short, otherwise it will be easily cracked by the rainbow table. Like the example I mentioned above, the length of fuzzy query characters must be greater than or equal to 4 letters/digits, or 2 Chinese characters.

2.5 Single-character digest algorithm (default fuzzy query algorithm provided in ShardingSphere version 5.3.0)

Although the above methods are all viable, it’s only natural to wonder if there’s a better alternative out there. In our community, we find that single-character encryption and storage can balance both performance and query, but fails to meet security requirements.

Then what’s the ideal solution? Inspired by masking algorithms and cryptographic hash functions, we find that data loss and one-way functions can be used.

The cryptographic hash function should have the following four features:

For any given message, it should be easy to calculate the hash value.
It should be difficult to infer the original message from a known hash value.
It should not be feasible to modify the message without changing the hash value.
There should only be a very low chance that two different messages produce the same hash value. Security: because of the one-way function, it’s not possible to infer the original message. In order to improve the accuracy of the fuzzy query, we want to encrypt a single character, but it will be cracked by the rainbow table.

So we take a one-way function (to make sure every character is the same after encryption) and increase the frequency of collisions (to make sure every string is 1: N backward), which greatly enhances security.

3. Fuzzy query algorithm

Apache ShardingSphere implements a universal fuzzy query algorithm by using the below single-character digest algorithm org.apache.shardingsphere.encrypt.algorithm.like.CharDigestLikeEncryptAlgorithm.

public final class CharDigestLikeEncryptAlgorithm implements LikeEncryptAlgorithm<Object, String> {

    private static final String DELTA = "delta";

    private static final String MASK = "mask";

    private static final String START = "start";

    private static final String DICT = "dict";

    private static final int DEFAULT_DELTA = 1;

    private static final int DEFAULT_MASK = 0b1111_0111_1101;

    private static final int DEFAULT_START = 0x4e00;

    private static final int MAX_NUMERIC_LETTER_CHAR = 255;

    @Getter
    private Properties props;

    private int delta;

    private int mask;

    private int start;

    private Map<Character, Integer> charIndexes;

    @Override
    public void init(final Properties props) {
        this.props = props;
        delta = createDelta(props);
        mask = createMask(props);
        start = createStart(props);
        charIndexes = createCharIndexes(props);
    }

    private int createDelta(final Properties props) {
        if (props.containsKey(DELTA)) {
            String delta = props.getProperty(DELTA);
            try {
                return Integer.parseInt(delta);
            } catch (NumberFormatException ex) {
                throw new EncryptAlgorithmInitializationException("CHAR_DIGEST_LIKE", "delta can only be a decimal number");
            }
        }
        return DEFAULT_DELTA;
    }

    private int createMask(final Properties props) {
        if (props.containsKey(MASK)) {
            String mask = props.getProperty(MASK);
            try {
                return Integer.parseInt(mask);
            } catch (NumberFormatException ex) {
                throw new EncryptAlgorithmInitializationException("CHAR_DIGEST_LIKE", "mask can only be a decimal number");
            }
        }
        return DEFAULT_MASK;
    }

    private int createStart(final Properties props) {
        if (props.containsKey(START)) {
            String start = props.getProperty(START);
            try {
                return Integer.parseInt(start);
            } catch (NumberFormatException ex) {
                throw new EncryptAlgorithmInitializationException("CHAR_DIGEST_LIKE", "start can only be a decimal number");
            }
        }
        return DEFAULT_START;
    }

    private Map<Character, Integer> createCharIndexes(final Properties props) {
        String dictContent = props.containsKey(DICT) && !Strings.isNullOrEmpty(props.getProperty(DICT)) ? props.getProperty(DICT) : initDefaultDict();
        Map<Character, Integer> result = new HashMap<>(dictContent.length(), 1);
        for (int index = 0; index < dictContent.length(); index++) {
            result.put(dictContent.charAt(index), index);
        }
        return result;
    }

    @SneakyThrows
    private String initDefaultDict() {
        InputStream inputStream = CharDigestLikeEncryptAlgorithm.class.getClassLoader().getResourceAsStream("algorithm/like/common_chinese_character.dict");
        LineProcessor<String> lineProcessor = new LineProcessor<String>() {

            private final StringBuilder builder = new StringBuilder();

            @Override
            public boolean processLine(final String line) {
                if (line.startsWith("#") || 0 == line.length()) {
                    return true;
                } else {
                    builder.append(line);
                    return false;
                }
            }

            @Override
            public String getResult() {
                return builder.toString();
            }
        };
        return CharStreams.readLines(new InputStreamReader(inputStream, Charsets.UTF_8), lineProcessor);
    }

    @Override
    public String encrypt(final Object plainValue, final EncryptContext encryptContext) {
        return null == plainValue ? null : digest(String.valueOf(plainValue));
    }

    private String digest(final String plainValue) {
        StringBuilder result = new StringBuilder(plainValue.length());
        for (char each : plainValue.toCharArray()) {
            char maskedChar = getMaskedChar(each);
            if ('%' == maskedChar) {
                result.append(each);
            } else {
                result.append(maskedChar);
            }
        }
        return result.toString();
    }

    private char getMaskedChar(final char originalChar) {
        if ('%' == originalChar) {
            return originalChar;
        }
        if (originalChar <= MAX_NUMERIC_LETTER_CHAR) {
            return (char) ((originalChar + delta) & mask);
        }
        if (charIndexes.containsKey(originalChar)) {
            return (char) (((charIndexes.get(originalChar) + delta) & mask) + start);
        }
        return (char) (((originalChar + delta) & mask) + start);
    }

    @Override
    public String getType() {
        return "CHAR_DIGEST_LIKE";
    }
}

Define the binary mask code to lose precision 0b1111_0111_1101 (mask).
Save common Chinese characters with disrupted order like a map dictionary.
Obtain a single string of Unicode for digits, English, and Latin.
Obtain index for a Chinese character belonging to a dictionary.
Other characters fetch the Unicode of a single string.
Add 1 (delta) to the digits obtained by different types above to prevent any original text from appearing in the database.
Then convert the offset Unicode into binary and perform the AND operation with mask, and carry out a 2-bit digit loss.
Directly output digits, English, and Latin after the loss of precision.
The remaining characters are converted to decimal and output with the common character start code after the loss of precision. ## 4. The fuzzy algorithm development progress ### 4.1 The first edition Simply use Unicode and mask code of common characters to perform the AND operation.

Mask: 0b11111111111001111101
The original character: 0b1000101110101111讯
After encryption: 0b1000101000101101設

Assuming we know the key and encryption algorithm, the original string after a backward pass is:
1.0b1000101100101101 謭
2.0b1000101100101111 謯
3.0b1000101110101101 训
4.0b1000101110101111 讯
5.0b1000101010101101 読
6.0b1000101010101111 誯
7.0b1000101000101111 訯
8.0b1000101000101101 設

We find that based on the missing bits, each string can be derived 2^n Chinese characters backward. When the Unicode of common Chinese characters is decimal, their intervals are very large. Notice that the Chinese characters inferred backward are not common characters, and it's more likely to infer the original characters.

4.2 The second edition

Since the interval of common Chinese characters Unicode is irregular, we planned to leave the last few bits of Chinese characters Unicode and convert them into decimal as index to fetch some common Chinese characters. This way, when the algorithm is known, uncommon characters won't appear after a backward pass, and distractors are no longer easy to eliminate.

If we leave the last few bits of Chinese characters Unicode, it has something to do with the relationship between the accuracy of fuzzy query and anti-decryption complexity. The higher the accuracy, the lower the decryption difficulty.

Let’s take a look at the collision degree of common Chinese characters under our algorithm:

When mask=0b0011_1111_1111:

When mask=0b0001_1111_1111:

For the mantissa of Chinese characters, leave 10 and 9 digits. The 10-digit query is more accurate because its collision is much weaker. Nevertheless, if the algorithm and the key are known, the original text of the 1:1 character can be derived backward.

The 9-digit query is less accurate because 9-digit collisions are relatively stronger, but there are fewer 1:1 characters. We find that although we change the collisions regardless of whether we leave 10 or 9 digits, the distribution is very unbalanced due to the irregular Unicode of Chinese characters, and the overall collision probability cannot be controlled.

4.3 The third edition

In response to the unevenly distributed problem found in the second edition, we take common characters with disrupted order as the dictionary table.

The encrypted text first looks up the index in the out-of-order dictionary table. We use the index and subscript to replace the Unicode without rules.
Use Unicode in case of uncommon characters. (Note: evenly distribute the code to be calculated as far as possible.)
The next step is to perform the AND operation with mask and lose 2-bit precision to increase the frequency of collisions.

Let’s take a look at the collision degree of common Chinese characters under our algorithm:

When mask=0b1111_1011_1101:

When mask=0b0111_1011_1101:

When the mask leaves 11 bits, you can see that the collision distribution is concentrated at 1:4. When mask leaves 10 bits, the number becomes 1:8. At this time, we only need to adjust the number of precision losses to control whether the collision is 1:2, 1:4 or 1:8.

If mask is selected as 1, and the algorithm and key are known, there will be a 1:1 Chinese character, because what we calculate at this time is the collision degree of common characters. If we add the missing 4 bits before the 16-bit binary of Chinese characters, the situation becomes 2^5=32 cases.

Since we encrypted the whole text, even if the individual character is inferred backwards, there will be little impact on overall security, and it will not cause mass data leaks. At the same time, the premise of backward pass is to know the algorithm, key, delta and dictionary, so it's impossible to achieve from the data in the database.

5. How to use fuzzy query

Fuzzy query requires the configuration of encryptors(encryption algorithm configuration), likeQueryColumn (fuzzy query column name), and likeQueryEncryptorName(encryption algorithm name of fuzzy query column ) in the encryption configuration.

Please refer to the following configuration. Add your own sharding algorithm and data source.

dataSources:
  ds_0:
    dataSourceClassName: com.zaxxer.hikari.HikariDataSource
    driverClassName: com.mysql.jdbc.Driver
    jdbcUrl: jdbc:mysql://127.0.0.1:3306/test?allowPublicKeyRetrieval=true
    username: root
    password: root

rules:
- !ENCRYPT
  encryptors:
    like_encryptor:
      type: CHAR_DIGEST_LIKE
    aes_encryptor:
      type: AES
      props:
        aes-key-value: 123456abc
  tables:
    user:
      columns:
        name:
          cipherColumn: name
          encryptorName: aes_encryptor
          assistedQueryColumn: name_ext
          assistedQueryEncryptorName: aes_encryptor
          likeQueryColumn: name_like
          likeQueryEncryptorName: like_encryptor
        phone:
          cipherColumn: phone
          encryptorName: aes_encryptor
          likeQueryColumn: phone_like
          likeQueryEncryptorName: like_encryptor
  queryWithCipherColumn: true


props:
  sql-show: true

Insert

Logic SQL: insert into user ( id, name, phone, sex) values ( 1, '熊高祥', '13012345678', '男')
Actual SQL: ds_0 ::: insert into user ( id, name, name_ext, name_like, phone, phone_like, sex) values (1, 'gyVPLyhIzDIZaWDwTl3n4g==', 'gyVPLyhIzDIZaWDwTl3n4g==', '佹堝偀', 'qEmE7xRzW0d7EotlOAt6ww==', '04101454589', '男')

Update

Logic SQL: update user set name = '熊高祥123', sex = '男1' where sex ='男' and phone like '130%'
Actual SQL: ds_0 ::: update user set name = 'K22HjufsPPy4rrf4PD046A==', name_ext = 'K22HjufsPPy4rrf4PD046A==', name_like = '佹堝偀014', sex = '男1' where sex ='男' and phone_like like '041%'

Select

Logic SQL: select * from user where (id = 1 or phone = '13012345678') and name like '熊%'
Actual SQL: ds_0 ::: select `user`.`id`, `user`.`name` AS `name`, `user`.`sex`, `user`.`phone` AS `phone`, `user`.`create_time` from user where (id = 1 or phone = 'qEmE7xRzW0d7EotlOAt6ww==') and name_like like '佹%'

Select: federated table sub-query

Logic SQL: select * from user LEFT JOIN user_ext on user.id=user_ext.id where user.id in (select id from user where sex = '男' and name like '熊%')
Actual SQL: ds_0 ::: select `user`.`id`, `user`.`name` AS `name`, `user`.`sex`, `user`.`phone` AS `phone`, `user`.`create_time`, `user_ext`.`id`, `user_ext`.`address` from user LEFT JOIN user_ext on user.id=user_ext.id where user.id in (select id from user where sex = '男' and name_like like '佹%')

Delete

Logic SQL: delete from user where sex = '男' and name like '熊%'
Actual SQL: ds_0 ::: delete from user where sex = '男' and name_like like '佹%'

The above example demonstrates how fuzzy query columns rewrite SQL in different SQL syntaxes to support fuzzy queries.

This blog post introduced you to the working principles of fuzzy query and used specific examples to demonstrate how to use it. We hope that through this article, you will have gained a basic understanding of fuzzy queries.

Links

🔗 Download Link

🔗 Project Address

🔗 ShardingSphere-on-Cloud

Author

Xiong Gaoxiang, an engineer at Iflytek and a ShardingSphere Contributor, is responsible for the data encryption and data masking R&D.

Use AWS CloudFormation to create ShardingSphere HA clusters

Apache ShardingSphere — Thu, 15 Dec 2022 08:37:52 +0000

What’s AWS CloudFormation?

AWS CloudFormation is an infrastructure as code (IaC) service that allows you to easily model, provision, and manage all the cloud
services provided by AWS by custom templates.

Traditionally, when we create a new architecture, we need to manually configure it in AWS step by step which can cause some errors such forgetting some steps.

Now with CloudFormation, we can use declarative configuration to define resources and then create and manage them without worrying about the dependency order of the resources.

In real-world scenarios, we often need to create duplicate architectures. For example, we can build a set of MQ clusters in a pre-release environment whose architecture and production environment are consistent, and create an AutoScalingGroup with the same configuration in each availability zone (AZ).

Through CloudFormation, those repeated architectures can then be expressed in the form of code and stored in the code repository for management. This way, they can be easily integrated into existing CI/CD pipelines to achieve the change of infrastructure, in accordance with the current DevOps process. As a result, the infrastructure change is more transparent, repeatable, testable and auditable, simplifying the management of cloud systems.

What’s Apache ShardingSphere?

Apache ShardingSphere is a distributed database ecosystem that can transform any database into a distributed database system, and enhance it with sharding, elastic scaling, encryption features & more.

Apache ShardingSphere follows the Database Plus concept, designed to build an ecosystem on top of fragmented heterogeneous databases. It focuses on how to fully use the computing and storage capabilities of databases rather than creating a brand-new database.

ShardingSphere-Proxy is a transparent database proxy that supports any client that uses MySQL, PostgreSQL and openGauss.

Deploy with CloudFormation

As an important part of the data infrastructure, the ShardingSphere Proxy cluster is significant for high availability. For these reasons, our community hopes that you can manage it through IaC to enjoy the benefits brought by IaC.

In the following sections we’ll use CloudFormation to create a ShardingSphere-Proxy cluster in multi-AZ. Before creating CloudFormation templates, we have to understand the architecture diagram of a ShardingSphere-Proxy cluster.

Please note that we use Zookeeper as the Governance Center.

As you can see, ShardingSphere-Proxy itself is a stateless application. In production scenarios, it only needs to offer a LoadBalancer, which can flexibly distribute the traffic between instances.

To ensure the HA of the ZooKeeper cluster and the ShardingSphere-Proxy cluster, we use the following architecture.

Define CloudFormation Parameters

The CloudFormation template is a yaml or json file in which you can define all of the infrastructure. The CloudFormation parameter allows you to inject custom values into the templates. You can then reference these parameters when defining resources. Of course, we also provide default values, which can be overridden if needed.

We use cfndsl to write CloudFormation templates: cfndsl is a tool written in ruby, allowing you to generate CloudFormation templates through DSL configuration.

Parameter("ZookeeperInstanceType") {
  String
  Default "t2.nano"
}

Parameter("ShardingSphereInstanceType") {
  String
  Default "t2.micro"
}

Parameter("KeyName") {
  String
  Default "test-tf"
  Description "The ssh keypair for remote connetcion"
}

Parameter("ImageId") {
  Type "AWS::EC2::Image::Id"
  Default "ami-094bbd9e922dc515d"
}

Parameter("VpcId") {
  String
  Default "vpc-0ef2b7440d3ade8d5"
  Description "The id of your VPC"
}

....

For more parameters, please refer to this link.

Define CloudFormation Resources

Note that AWS’s region usually includes more than two AZs, so we created three instances.

Zookeeper Clusters

First, we deploy a Zookeeper EC2 instance in each AZ to ensure the HA of the ZooKeeper cluster.

Then we create an internal domain name for each Zookeeper instance, and ShardingSphere-Proxy uses these domain names.

When the ZooKeeer instance starts, cloud-init is used to automatically deploy the Zookeeper service. You can view the cloud-init configuration here.

(0..2).each do |i| 
    name = "ZK#{i+1}"
    EC2_Instance(name) {
      AvailabilityZone FnSelect(i, FnGetAZs(Ref("AWS::Region")))
      InstanceType Ref("ZookeeperInstanceType")
      ImageId Ref("ImageId")
      KeyName Ref("KeyName")
      SubnetId FnSelect(i, Ref("Subnets"))
      SecurityGroupIds Ref("SecurityGroupIds")
      Tags [ 
        Tag do 
          Key "Name"
          Value "ZK-#{i+1}"
        end
      ]

      server = "server.%{idx}=zk-%{idx}.${HostedZoneName}:2888:3888"
      UserData FnBase64(
        FnSub(
          IO.read("./zookeeper-cloud-init.yml"), 
          :SERVERS => FnSub((0..2).map{|i| i == 0 ? server %{:idx => i+1} : ("#{server}" %{:idx => i+1}).insert(0, " " * 4)}.join("\n")), 
          :VERSION => Ref("ZookeeperVersion"),
          :ZK_HEAP => Ref("ZookeeperHeap"),
          :INDEX => i+1,
        )
      )
    }

    domain = "zone#{name}"
    Route53_RecordSet(domain) {
      HostedZoneId Ref("HostedZoneId")
      Name FnSub("zk-#{i+1}.${HostedZoneName}")
      Type "A"
      ResourceRecords [FnGetAtt(name, "PrivateIp")]
      TTL "60"
    }
  end

ShardingSphere-Proxy Clusters

LaunchTemplate

Next, we deploy an AutoScalingGroup in each AZ to ensure the HA of a ShardingSphere-Proxy cluster.

Before creating the AutoScalingGroup, we need to create a LaunchTemplate in each AZ for the ShardingSphere-Proxy instance.

Similarly, when the instance starts, cloud-init is used to automatically deploy the ShardingSphere-Proxy service. You can view the cloud-init configuration here.

(0..2).each do |i| 
    name = "launchtemplate#{i}"
    EC2_LaunchTemplate(name) {
      LaunchTemplateName FnSub("shardingsphere-${TMPL_NAME}", :TMPL_NAME => FnSelect(i, FnGetAZs(Ref('AWS::Region'))))
      LaunchTemplateData do 
        ImageId Ref("ImageId")
        InstanceType Ref("ShardingSphereInstanceType")
        KeyName Ref("KeyName")

        MetadataOptions do
          HttpEndpoint "enabled"
          HttpTokens   "required"
          InstanceMetadataTags "enabled"
        end

        Monitoring do
          Enabled  true
        end

        NetworkInterfaces [
          {
            :DeleteOnTermination => false,
            :DeviceIndex => 0,
            :NetworkInterfaceId => FnGetAtt("networkiface#{i}", "Id")
          }
        ]

        TagSpecifications [
          {
            :ResourceType => "instance",
            :Tags => [
              {
                :Key => "Name",
                :Value => "shardingsphere-#{i+1}"
              }
            ]
          }
        ]

        UserData FnBase64(
          FnSub(
            IO.read("./shardingsphere-cloud-init.yml"), 
            :ZK_SERVERS => FnSub((0..2).map{|i| "zk-#{i+1}.${HostedZoneName}:2181" }.join(",")), 
            :VERSION => Ref("ShardingSphereVersion"),
            :JAVA_MEM_OPTS => Ref("ShardingSphereJavaMemOpts")
          )
        )
      end
    }
  end

TargetGroup

As we use ELB to load traffic between each ShardingSphere-Proxy, ELB should be used in combination with TargetGroup.

Instances created by AutoScalingGroup are automatically registered to TargetGroup. ELB will then forward the traffic to TargetGroup.

ElasticLoadBalancingV2_TargetGroup("sslbtg") {
    Name "shardingsphere-lb-tg"
    Port Ref("ShardingSpherePort")
    Protocol "TCP"
    VpcId Ref("VpcId")
    TargetGroupAttributes [
      TargetGroupAttribute do
        Key "preserve_client_ip.enabled"
        Value "false"
      end
    ]
    Tags [
      Tag do
        Key "Name"
        Value "shardingsphere"
      end
    ]
  }

AutoScalingGroup

After creating the aforementioned resources, now we can create an AutoScalingGroup.

(0..2).each do |i| 
    name = "autoscaling#{i}"
    AutoScaling_AutoScalingGroup(name) {
      AutoScalingGroupName "shardingsphere-#{i}" 
      AvailabilityZones [FnSelect(i, FnGetAZs(Ref("AWS::Region")))]
      DesiredCapacity "1"
      MaxSize "1"
      MinSize "1"
      HealthCheckGracePeriod  60
      HealthCheckType "EC2"

      TargetGroupARNs [ Ref("sslbtg")]

      LaunchTemplate do
        LaunchTemplateName  FnSub("shardingsphere-${TMPL_NAME}", :TMPL_NAME => FnSelect(i, FnGetAZs(Ref('AWS::Region'))))
        Version FnGetAtt("launchtemplate#{i}", "LatestVersionNumber")
      end
    }
  end

LoadBalancer & Listener

Create an internal LoadBalancer and Listener for the external service of the ShardingSphere-Proxy cluster.

ElasticLoadBalancingV2_LoadBalancer("ssinternallb") {
    Name "shardingsphere-internal-lb"
    Scheme "internal"
    Type "network"

    mappings = (0..2).map { |x| 
        SubnetMapping do
          SubnetId FnSelect(x, Ref("Subnets"))
        end
    }
    SubnetMappings mappings
    Tags [
      Tag do
        Key "Name"
        Value "shardingsphere"
      end
    ]
  }

  ElasticLoadBalancingV2_Listener("sslblistener") {
    Port Ref("ShardingSpherePort")
    LoadBalancerArn Ref("ssinternallb")
    Protocol "TCP"
    DefaultActions [
      {
        :Type => "forward",
        :TargetGroupArn => Ref("sslbtg")
      }
    ]
  }

Internal Domain Names

Finally, we create the internal domain names for the external service of the ShardingSphere-Prxoy cluster. Those domain names point to the internal LoadBalancer.

Route53_RecordSet("ssinternaldomain") {
    HostedZoneId Ref("HostedZoneId")
    Name FnSub("proxy.${HostedZoneName}")
    Type "A"
    AliasTarget do 
      HostedZoneId FnGetAtt("ssinternallb", "CanonicalHostedZoneID")
      DNSName FnGetAtt("ssinternallb", "DNSName")
      EvaluateTargetHealth true
    end
  }

Deployment

Use the command cfndsl cf.rb -o cf.json --pretty to generate the final configuration.

Create Stack on the UI page and select the config file we generated.

After a few minutes, you will find that all the resources have been created.

You can find the complete code here, or visit our website for more information.

Test

The test is designed to ensure that the clusters we created are workable. A simple case is illustrated below.

Use DistSQL (Distributed SQL) to add two data sources and create a simple sharding rule, and then insert data. We can see that the query returns correct results.

By default, when we use CloudFormation, an internal domain name proxy.shardingsphere.org will be created. The username and password of the ShardingSphere-Proxy cluster are both root.

Note: DistSQL (Distributed SQL) is ShardingSphere’s SQL-like operating language. It’s used the same way as standard SQL, and is designed to provide incremental SQL operation capability.

Conclusion

AWS CloudFormation is an incredibly powerful service that is really helpful for the iteration of ShardingSphere-Proxy clusters. With this new addition, it is now easier than ever to get started with Apache ShardingSphere.

ShardingSphere 5.3.0 is released: new features and improvements

Apache ShardingSphere — Mon, 12 Dec 2022 07:31:39 +0000

After 1.5 months in development, Apache ShardingSphere 5.3.0 is released. Our community merged 687 PRs from contributors around the world.

The new release has been improved in terms of features, performance, testing, documentation, examples, etc.

The 5.3.0 release brings the following highlights:

Support fuzzy query for CipherColumn.
Support Datasource-level heterogeneous database.
Support checkpoint resume for data consistency check.
Automatically start a distributed transaction, while executing DML statements across multiple shards.

Additionally, release 5.3.0 also brings the following adjustments:

Remove the Spring configuration.
Systematically refactor the DistSQL syntax.
Refactor the configuration format of ShardingSphere-Proxy.

Highlights

1.Support fuzzy query for CipherColumn

In previous versions, ShardingSphere’s Encrypt feature didn’t support the use of the LIKE operator in SQL.

For a while users strongly requested adding the LIKE operator to the Encrypt feature. Usually, encrypted fields are mainly of the string type, and it is a common practice for the string to execute LIKE.

To minimize friction in accessing the Encrypt feature, our community has initiated a discussion about the implementation of encrypted LIKE.

Since then, we’ve received a lot of feedback.

Some community members even contributed their original encryption algorithm implementation supporting fuzzy queries after fully investigating conventional solutions.

🔗 The relevant issue can be found here.

🔗 For the algorithm design, please refer to the attachment within the issue.

The [single-character abstract algorithm] contributed by the community members is implemented as CHAR_DIGEST_LIKE in the ShardingSphere encryption algorithm SPI.

2. Support datasource-level heterogeneous database

ShardingSphere supports a database gateway, but its heterogeneous capability is limited to the logical database in previous versions. This means that all the data sources under a logical database must be of the same database type.

This new release supports datasource-level heterogeneous databases at the kernel level. This means the datasources under a logical database can be different database types, allowing you to use different databases to store data.

Combined with ShardingSphere’s SQL dialect conversion capability, this new feature significantly enhances ShardingSphere’s heterogeneous data gateway capability.

3. Data migration: support checkpoint resume for data consistency check

Data consistency checks happen at the later stage of data migration.

Previously, the data consistency check was triggered and stopped by DistSQL. If a large amount of data was migrated and the data consistency check was stopped for any reason, the check would’ve had to be started again — which is sub-optimal and affects user experience.

ShardingSphere 5.3.0 now supports checkpoint storage, which means data consistency checks can be resumed from the checkpoint.

For example, if data is being verified during data migration and the user stops the verification for some reason, with the verification progress (finished_percentage) being 5%, then:

mysql> STOP MIGRATION CHECK 'j0101395cd93b2cfc189f29958b8a0342e882';
Query OK, 0 rows affected (0.12 sec)

mysql> SHOW MIGRATION CHECK STATUS 'j0101395cd93b2cfc189f29958b8a0342e882';
+--------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| tables | result | finished_percentage | remaining_seconds | check_begin_time        | check_end_time          | duration_seconds | error_message |
+--------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
| sbtest | false  | 5                   | 324               | 2022-11-10 19:27:15.919 | 2022-11-10 19:27:35.358 | 19               |               |
+--------+--------+---------------------+-------------------+-------------------------+-------------------------+------------------+---------------+
1 row in set (0.02 sec)

In this case, the user restarts the data verification. But the work does not have to restart from the beginning. The verification progress (finished_percentage) remains at 5%.

mysql> START MIGRATION CHECK 'j0101395cd93b2cfc189f29958b8a0342e882';
Query OK, 0 rows affected (0.35 sec)

mysql> SHOW MIGRATION CHECK STATUS 'j0101395cd93b2cfc189f29958b8a0342e882';
+--------+--------+---------------------+-------------------+-------------------------+----------------+------------------+---------------+
| tables | result | finished_percentage | remaining_seconds | check_begin_time        | check_end_time | duration_seconds | error_message |
+--------+--------+---------------------+-------------------+-------------------------+----------------+------------------+---------------+
| sbtest | false  | 5                   | 20                | 2022-11-10 19:28:49.422 |                | 1                |               |
+--------+--------+---------------------+-------------------+-------------------------+----------------+------------------+---------------+
1 row in set (0.02 sec)

Limitation: this new feature is unavailable with the CRC32_MATCH algorithm because the algorithm calculates all data at once.

4. Automatically start a distributed transaction while executing DML statements across multiple shards

Previously, even with XA and other distributed transactions configured, ShardingSphere could not guarantee the atomicity of DML statements that are routed to multiple shards — if users didn’t manually enable the transaction.

Take the following SQL as an example:

insert into account(id, balance, transaction_id) values
(1, 1, 1),(2, 2, 2),(3, 3, 3),(4, 4, 4),
(5, 5, 5),(6, 6, 6),(7, 7, 7),(8, 8, 8);

When this SQL is sharded according to id mod 2, the ShardingSphere kernel layer will automatically split it into the following two SQLs and route them to different shards respectively for execution:

insert into account(id, balance, transaction_id) values
(1, 1, 1),(3, 3, 3),(5, 5, 5),(7, 7, 7);

insert into account(id, balance, transaction_id) values
(2, 2, 2),(4, 4, 4),(6, 6, 6),(8, 8, 8);

If the user does not manually start the transaction, and one of the sharded SQL fails to execute, the atomicity cannot be guaranteed because the successful operation cannot be rolled back.

ShardingSphere 5.3.0 is optimized in terms of distributed transactions. If distributed transactions are configured in ShardingSphere, they can be automatically started when DML statements are routed to multiple shards. This way, we can ensure atomicity when executing DML statements.

Significant Adjustments

1.Remove Spring configuration

In earlier versions, ShardingSphere-JDBC provided services in the format of DataSource. If you wanted to introduce ShardingSphere-JDBC without modifying the code in the Spring/Spring Boot project, you needed to use modules such as Spring/Spring Boot Starter provided by ShardingSphere.

Although ShardingSphere supports multiple configuration formats, it also has the following problems:

When API changes, many config files need to be adjusted, which is a heavy workload.
The community has to maintain multiple config files.
The lifecycle management of Spring bean is susceptible to other dependencies of the project such as PostProcessor failure.
Spring Boot Starter and Spring NameSpace are affected by Spring, and their configuration styles are different from YAML.
Spring Boot Starter and Spring NameSpace are affected by the version of Spring. When users access them, the configuration may not be identified and dependency conflicts may occur. For example, Spring Boot versions 1.x and 2.x have different configuration styles.

ShardingSphere 5.1.2 first supported the introduction of ShardingSphere-JDBC in the form of JDBC Driver. That means applications only need to configure the Driver provided by ShardingSphere at the JDBC URL before accessing to ShardingSphere-JDBC.

Removing the Spring configuration simplifies and unifies the configuration mode of ShardingSphere. This adjustment not only simplifies the configuraiton of ShardingSphere when using different configuration modes, but also reduces maintenance work for the ShardingSphere community.

2. Systematically refactor the DistSQL syntax

One of the characteristics of Apache ShardingSphere is its flexible rule configuration and resource control capability.

DistSQL (Distributed SQL) is ShardingSphere’s SQL-like operating language. It’s used the same way as standard SQL, and is designed to provide incremental SQL operation capability.

ShardingSphere 5.3.0 systematically refactors DistSQL. The community redesigned the syntax, semantics and operating procedure of DistSQL. The new version is more consistent with ShardingSphere’s design philosophy and focuses on a better user experience.

Please refer to the latest ShardingSphere documentation for details. A DistSQL roadmap will be available soon, and you’re welcome to leave your feedback.

3. Refactor the configuration format of ShardingSphere-Proxy

In this update, ShardingSphere-Proxy has adjusted the configuration format and reduced config files required for startup.

server.yaml before refactoring:

rules:
  - !AUTHORITY
    users:
      - root@%:root
      - sharding@:sharding
    provider:
      type: ALL_PERMITTED
  - !TRANSACTION
    defaultType: XA
    providerType: Atomikos
  - !SQL_PARSER
    sqlCommentParseEnabled: true
    sqlStatementCache:
      initialCapacity: 2000
      maximumSize: 65535
    parseTreeCache:
      initialCapacity: 128
      maximumSize: 1024

server.yaml after refactoring:

authority:
  users:
    - user: root@%
      password: root
    - user: sharding
      password: sharding
  privilege:
    type: ALL_PERMITTED

transaction:
  defaultType: XA
  providerType: Atomikos
sqlParser:
  sqlCommentParseEnabled: true
  sqlStatementCache:
    initialCapacity: 2000
    maximumSize: 65535
  parseTreeCache:
    initialCapacity: 128
    maximumSize: 1024

In ShardingSphere 5.3.0, server.yaml is no longer required to start Proxy. If no config file is provided by default, Proxy provides the default account root/root.

ShardingSphere is completely committed to becoming cloud native. Thanks to DistSQL, ShardingSphere-Proxy’s config files can be further simplified, which is more friendly to container deployment.

Release Notes

API Changes

DistSQL: refactor syntax API, please refer to the user manual
Proxy: change the configuration style of global rule, remove the exclamation mark
Proxy: allow zero-configuration startup, enable the default account root/root when there is no Authority configuration
Proxy: remove the default logback.xml and use API initialization
JDBC: remove the Spring configuration and use Driver + YAML configuration instead

Enhancements

DistSQL: new syntax REFRESH DATABASE METADATA, refresh logic database metadata
Kernel: support DistSQL REFRESH DATABASE METADATA to load configuration from the governance center and rebuild MetaDataContext
Support PostgreSQL/openGauss setting transaction isolation level
Scaling: increase inventory task progress update frequence
Scaling: DATA_MATCH consistency check support checkpoint resume
Scaling: support drop consistency check job via DistSQL
Scaling: rename column from sharding_total_count to job_item_count in job list DistSQL response
Scaling: add sharding column in incremental task SQL to avoid broadcast routing
Scaling: sharding column could be updated when generating SQL
Scaling: improve column value reader for DATA_MATCH consistency check
DistSQL: encrypt DistSQL syntax optimization, support like query algorithm
DistSQL: add properties value check when REGISTER STORAGE UNIT
DistSQL: remove useless algorithms at the same time when DROP RULE
DistSQL: EXPORT DATABASE CONFIGURATION supports broadcast tables
DistSQL: REGISTER STORAGE UNIT supports heterogeneous data sources
Encrypt: support Encrypt LIKE feature
Automatically start distributed transactions when executing DML statements across multiple shards
Kernel: support client \d for PostgreSQL and openGauss
Kernel: support select group by, order by statement when column contains null values
Kernel: support parse RETURNING clause of PostgreSQL/openGauss Insert
Kernel: SQL HINT performance improvement
Kernel: support mysql case when then statement parse
Kernel: support data source level heterogeneous database gateway
(Experimental) Sharding: add sharding cache plugin
Proxy: support more PostgreSQL datetime formats
Proxy: support MySQL COM_RESET_CONNECTION
Scaling: improve MySQLBinlogEventType.valueOf to support unknown event type
Kernel: support case when for federation

Bug Fix

Scaling: fix barrier node created at job deletion
Scaling: fix part of columns value might be ignored in DATA_MATCH consistency check
Scaling: fix jdbc url parameters are not updated in consistency check
Scaling: fix tables sharding algorithm type INLINE is case-sensitive
Scaling: fix incremental task on MySQL require mysql system database permission
Proxy: fix the NPE when executing select SQL without storage node
Proxy: support DATABASE_PERMITTED permission verification in unicast scenarios
Kernel: fix the wrong value of worker-id in show compute nodes
Kernel: fix route error when the number of readable data sources and weight configurations of the Weight algorithm are not equal
Kernel: fix multiple groups of readwrite-splitting refer to the same load balancer name, and the load balancer fails problem
Kernel: fix can not disable and enable compute node problem
JDBC: fix data source is closed in ShardingSphereDriver cluster mode when startup problem
Kernel: fix wrong rewrite result when part of logical table name of the binding table is consistent with the actual table name, and some are inconsistent
Kernel: fix startup exception when use SpringBoot without configuring rules
Encrypt: fix null pointer exception when Encrypt value is null
Kernel: fix oracle parsing does not support varchar2 specified type
Kernel: fix serial flag judgment error within the transaction
Kernel: fix cursor fetch error caused by wasNull change
Kernel: fix alter transaction rule error when refresh metadata
Encrypt: fix EncryptRule cast to TransparentRule exception that occurs when the call procedure statement is executed in the Encrypt scenario
Encrypt: fix exception which caused by ExpressionProjection in shorthand projection
Proxy: fix PostgreSQL Proxy int2 negative value decoding incorrect
Proxy: PostgreSQL/openGauss support describe insert returning clause
Proxy: fix gsql 3.0 may be stuck when connecting Proxy
Proxy: fix parameters are missed when checking SQL in Proxy backend
Proxy: enable MySQL Proxy to encode large packets
Kernel: fix oracle parse comment without whitespace error
DistSQL: fix show create table for encrypt table

Refactor

Scaling: reverse table name and column name when generating SQL if it’s SQL keyword
Scaling: improve incremental task failure handling
Kernel: governance center node adjustment, unified hump to underscore

Links

🔗 Download Link

🔗 Release Notes

🔗 Project Address

🔗 ShardingSphere-on-Cloud

Community Contribution

This Apache ShardingSphere 5.3.0 release is the result of 687 merged PRs, committed by 49 contributors. Thank you for your efforts.

ShardingSphere-on-Cloud & Pisanix replace Sidecar for a true cloud-native experience

Apache ShardingSphere — Fri, 25 Nov 2022 02:55:06 +0000

Background

For a while, many of our blog posts have shown that ShardingSphere consists of three independent products: ShardingSphere-JDBC, ShardingSphere-Proxy, and ShardingSphere-Sidecar.

As ShardingSphere has become increasingly popular, JDBC and Proxy have been used in numerous production environments, but Sidecar’s status has remaing as “under planning”. you may have noticed this on our GitHub READ ME or our website.

With cloud-native gaining momentum, many enterprises are choosing to use databases on the cloud or cloud native databases. This represents an excellent opportunity for ShardingSphere-Sidecar, which is positioned as a cloud-native database proxy in Kubernetes.

However, some of you may have noticed that any mention of Sidecar has disappeared from ShardingSphere’s latest documentation. Has ShardingSphere-Sidecar been canceled? What will ShardingSphere’s cloud native future look like? Here’s what’s coming.

What’s ShardingSphere’s plan for the cloud?

Stop the R&D of ShardingSphere-Sidecar

As far as ShardingSphere users are concerned, ShardingSphere-JDBC and ShardingSphere-Proxy can already meet most requirements. ShardingSphere-Sidecar’s only difference from them is in the deployment form. JDBC and Proxy are functionally identical, but each has unique advantages.

ShardingSphere-JDBC is positioned as a lightweight Java framework that provides additional services at theJava JDBC layer. With the client connecting directly to the database, it provides services in the form of jar and requires no extra deployment and dependence. It can be viewed as an enhanced JDBC driver, fully compatible with JDBC and all kinds of ORM frameworks. It’s targeted at developers and boasts higher performance.

ShardingSphere-Proxy is a transparent database proxy, supporting heterogeneous languages through the implementation of database binary protocol. Currently, MySQL and PostgreSQL protocols are provided. Nevertheless, Proxy adds a gateway between the database and the front-end application layer, which will partially lower the performance.

Our community recommends users to adopt a hybrid deployment mode allowing JDBC and Proxy to complement each other, maximizing ShardingSphere’s advantages in terms of performance, availability and heterogeneous support.

As you can see, it’s a bit awkward for ShardingSphere-Sidecar: JDBC and Proxy are applicable to most scenarios, environments and businesses, and can complement each other, leaving little room for Sidecar to innovate. From the viewpoint of the community and its users, Sidecar is more like an extension in deployment mode, and it is not capable of enhancing ShardingSphere as a whole.

Taking the above factors into consideration, it is more efficient to develop a patch tool for ShardingSphere that can be easily used and run in a Kubernetes environment. This way, users can deploy and use ShardingSphere in cloud native environments, while saving R&D time for the ShardingSphere community.

ShardingSphere’s cloud solution: ShardingSphere-On-Cloud

ShardingSphere-on-Cloud is a comprehensive system upgrade based on ShardingSphere.

ShardingSphere-Sidecar was born when Kubernetes thrived. Back then, more and more enterprises were trying to adopt cloud-native concepts. The ShardingSphere community is not an exception.

We proposed ShardingSphere-Sidecar to promote cloud-native transformation in the data field. However, since JDBC and Proxy are mature enough to deal with data governance in most scenarios, it’s unnecessary to make ShardingSphere entirely cloud native.

Sidecar can indeed play a big role in certain scenarios, but it doesn’t mean that we have to create a Sidecar version for each component. ShardingSphere is working on how to come up with a solution based on real cloud-native scenarios after fully integrating the cloud computing concept. That’s how ShardingSphere-on-Cloud was born.

ShardingSphere-on-Cloud is capable of deploying and migrating ShardingSphere in a Kubernetes environment. With the help of AWS CloudFormation, Helm, Operator, and Terraform (coming soon) and other tools, it provides best practices with quick deployment, higher observability, security and migration, and high availability deployment in a cloud native environment.

Please refer to Database Plus Embracing the Cloud: ShardingSphere-on-Cloud Solution Released for details.

Achieving the vision of ShardingSphere-Sidecar through Pisanix

Why did we develop a new open source project oriented towards data governance in cloud native scenarios?

Our community has been contemplating the position of ShardingSphere and Database Mesh concept.

Within the community we hold different viewpoints on Sidecar at different stages. In the beginning, the community wanted to use Sidecar to manage cloud data issues. As the community gained a deeper understanding of cloud native and cloud data management processes, the limitations of ShardingSphere-Sidecar have been gradually exposed.

ShardingSphere-Sidecar is only a deployment mode of ShardingSphere in cloud native environments, so it can only solve a single problem. It is incapable of helping ShardingSphere develop a mature and cloud-native solution for enterprises.

Therefore, we needed to redesign an open source product with higher adaptability, availability and agility in a cloud native system - in order to make up for ShardingSphere’s limitations on cloud data governance.

That is why some of our community members at SphereEx developed Pisanix, a cloud-native data governance tool, based on the Database Mesh concept. It can provide capabilities such as SQL-aware traffic governance, runtime resource-oriented management and DBRE.

Is Pisanix the same with ShardingSphere-Sidecar?

ShardingSphere-Sidecar and Pisanix convey different understandings of Database Mesh. They are different in the following aspects.

Different design concepts: the design philosophy of JDBC & Proxy is Database Plus, which adds an interface on top of multiple data sources for unified governance and enhancement. Pisanix represents the specific practice of Database Mesh concept, leading to efficient and smooth DevOps of databases for applications in cloud native scenarios.
Different language ecosystems: JDBC is developed solely in Java. As Java is popular with numerous enterprise-grade community users and developers, JDBC can be easily applied to Java-developed applications. In comparison, Pisanix is developed in Rust to improve the reliability and efficiency of the access layer. Despite small differences, both of them are oriented towards cloud-native data infrastructure. That is also what Database Mesh expects in the long term by implementing cloud native DBRE.

In terms of deployment mode, Pisanix and ShardingSphere-Sidecar can both be deployed with business applications in the form of Sidecar, providing standard protocol access for developers. Moreover, Pisanix is highly compatible with the ShardingSphere ecosystem. You can connect Pisanix to ShardingSphere-Proxy in the same way that ShardingSphere-Proxy connects to MySQL.

In short, ShardingSphere presents a complete database form for developers and applications under the Database Plus concept. Pisanix is designed for the same purpose. Through Pisanix, the entrance to cloud data traffic, users can use ShardingSphere-Proxy as a database and explore the collaboration mode in a cloud native environment.

However, they belong to independent product lines. Pisanix followed the Database Mesh concept from the very beginning and achieved high-performance expansion through four aspects, including local database, unified configuration and management, multi-protocol support and cloud native architecture.

Pisanix is only the first step towards unifying database types on the cloud, and Sidecar is only a deployment form.

The protocols and DevOps features of different databases vary, and the point lies in abstracting the standard governance behavior.

Unlike Pisanix, the ShardingSphere ecosystem is not only accessible through protocols, and ShardingSphere-JDBC can be used more conveniently by Java applications.

Its natural compatibility maintains functional integrity, optimizes resource utilization and provides ultimate performance, with which developers can configure data governance at their will from the perspective of businesses. Meanwhile, by combining ShardingSphere and the underlying databases, users can deploy strong computing capability on the application side and transform the original monolithic database into a distributed database with high performance, optimizing resource allocation and offering a cost-effective solution.

In conclusion, ShardingSphere and Pisanix together offer two solutions for community users. **For users who’d like to deploy ShardingSphere in a Kubernetes environment, ShardingSphere-on-Cloud is enough, and ShardingSphere’s other features are exactly the same as when used locally.

For users looking to achieve unified traffic governance on the upper-layer database in cloud native scenarios, Pisanix is a better choice.**

Compared with ShardingSphere-Sidecar, ShardingSphere-on-Cloud combined with Pisanix is more effective and convenient.

ShardingSphere Official Website

Database Mesh Official Website

Pisanix Official Website

ElasticJob UI now supports Auth 2.0, OIDC and SAML single sign-on thanks to Casdoor

Apache ShardingSphere — Thu, 24 Nov 2022 03:12:42 +0000

If you‘re looking to add SSO to the administration console when using ElasticJob UI, this article will help you tackle this user management problem using ElasticJob UI’s built-in Casdoor.

Background

ElasticJob UI is the visual admin console of ElasticJob, whose target users are developers and DevOps teams rather than users. Generally, it is deployed only in the internal environment and thus its R&D focus more on its features.

Any open source project will inevitably become an object of interest for security researchers. Previously, security researchers submitted a lot of security issues on ElasticJob UI to the Apache ShardingSphere community without considering its actual usage scenario.

After careful considerations regarding dealing with those security vulnerability reports, the community decided to stop maintaining ElasticJob UI.

The Casdoor community members noticed our discussion about stopping the maintenance of ElasticJob UI and thought it would be a loss to deactivate ElasticJob UI because of unrealistic security reports.

If ElasticJob UI would be connected to Casdoor, it would be improved in terms of authentication security and features. In this context, the Casdoor and Apache ShardingSphere community reached a consensus on collaboration.

How to connect ElasticJob UI to Casdoor?

Step 1: Deploy Casdoor

The source code of Casdoor is on GitHub, and its boot mode includes development mode and production mode. The development mode is taken as an example here. Please refer to this link for more details.

Backend boot mode

go run main.go

Front-end boot mode

cd web
yarn install
yarn start

Step 2: Configure Casdoor and obtain the required data

The red arrows indicate what the backend configuration requires, with “Redirect URLs” referring to the address where you perform a callback.

We also need to find the corresponding cert we selected in the cert option, such as cert-build-in here. A certificate is also needed.

For additional Casdoor documentation, please refer to this link.

Step 3: Configuration in ElasticJob UI

Find application.properties in shardingsphere-elasticjob-ui and configure it.

Paste the data we obtained from Casdoor into the corresponding position as follows:

Now, we can use Casdoor in ElasticJob UI.

Once the ElasticJob’s admin console connects to Casdoor, it will support UI-first centralized identity access/single sign-on based on OAuth 2.0, OIDC and SAML.

Thanks to developers from the Casdoor and Apache ShardingSphere community, our collaboration has been going on in a smooth and friendly way. At first, jakiuncle from Casdoor proposed an issue and committed a PR, and then our Committer TeslaCN and PMC tristaZero reviewed the PR. This cross-community collaboration stands as a testament to the Beaty of open source.

About ElasticJob UI

ElasticJob is a distributed scheduling solution oriented towards Internet applications and massive tasks.

It provides elastic scheduling, resource management, and job governance combined with open architecture design, building a diversified job ecosystem. It uses a unified job API for each project.

ElasticJob-UI is the visual admin console of ElasticJob, supporting dynamic configuration, job management and control, job history retrieval and other features.

🔗 GitHub

About Casdoor

Casdoor is a UI-first identity access management (IAM) / single-sign-on (SSO) platform based on OAuth 2.0 / OIDC. Casdoor can help you solve user management problems. There’s no need to develop a series of authentication features such as user login and registration. It can manage the user module entirely in a few simple steps in conjunction with the host application. It’s convenient, easy-to-use and powerful.

🔗 GitHub

Alibaba Cloud’s OpenSergo & ShardingSphere release database governance standard for microservices

Apache ShardingSphere — Thu, 17 Nov 2022 03:55:15 +0000

Background

Recently, Alibaba Cloud’s OpenSergo and ShardingSphere jointly released the database governance standard for microservices. By combining the Database Plus and Database Mesh concepts, the two
communities have standardized the existing database governance concepts, patterns, and paths, further completing the database governance ecosystem under the cloud-native environment.

The founders of both communities expressed their opinions concerning the collaboration between the ShardingSphere community and the OpenSergo community:

Zhang Liang, the PMC Chair of the Apache ShardingSphere community:

in the microservices field, the interaction and collaboration between services have been gradually perfected, but there is still no effective standard for services to access the database. Being, ShardingSphere has been continuously following the “connect, enhance, and pluggable” design philosophy. “Connect” refers to providing standardized protocols and interfaces, breaking the barriers for development languages to access heterogeneous databases. It’s forward-looking for OpenSergo to propose the microservice-oriented governance standard and take the initiative to include database access in the standard. I’m pleased to see ShardingSphere and OpenSergo working together to build the standard for microservices which is a pivotal entrance to access databases.

Zhao Yihao, the founder of the OpenSergo community:

in microservice governance, in addition to the governance of the microservices itself, it is also a critical step to ensure business reliability and continuity to deal with microservices’ access to databases. As a Top-Level project in the database governance field, ShardingSphere has integrated a wealth of best practices and technical experience, which complements OpenSergo. In this context, we work with the ShardingSphere community to jointly build a database governance standard from the perspective of microservices, so that enterprises can carry out unified and standardized governance and management on different data layer frameworks and traffic.

Note: database governance in this article includes all aspects of database governance in microservice systems. All business information and key data need a robust and stable database system as it is the most important state terminal.

The database is increasingly vital in the microservice system

To meet flexible business requirements, the application architecture can be transformed from monolithic to service-oriented and then to microservice-oriented. In this case, the database used to store the core data becomes the focus of the distributed system.

Enterprises take advantage of microservices to separate services and adopt a distributed architecture to achieve loose coupling among multiple services, flexible adjustment and combination of services, and high availability of systems. In particular, microservices indeed have delivered enormous benefits in the face of rapidly changing businesses.

However, after services are separated, their corresponding underlying databases should also be separated to ensure the independent deployment of each service. In this way, each service can be an independent unit, finally achieving the microservices. In this context, the database is becoming increasingly complicated:

The transformation from monolithic to microservice-oriented architecture models leads to increasingly complicated services, diversified requirements, larger scale infrastructure, complex call relations between services, and higher requirements on the underlying database performance.
Different transactions usually involve multiple services — but it is a challenge to ensure data consistency between services.
It is also challenging to query data across multiple services.

With most backend applications, their system performance improvement is mainly limited to databases. Particularly in a microservice environment, it is a team’s top priority to deal with database performance governance. Database governance naturally becomes an indispensable part of microservice governance.

In database governance, we now mainly focus on read/write splitting, sharding, shadow databases, database discovery, and distributed transactions. At the same time, how to use databases and the actual database storage nodes rely on the virtual database and database endpoint.

In response to the above-mentioned problems, OpenSergo and ShardingSphere have assimilated the latter’s database governance experience and released the database governance standard under microservices. By doing so, they can standardize the database governance method, lower the entry barriers of this field, and improve the business applicability of microservices.

ShardingSphere’s strategies on database traffic governance

1. VirtualDatabase

In database governance, whether it is read/write splitting, sharding, shadow database or encryption, audit, and access control, they all have to act on a specific database. Here such a logical database is called a virtual database, namely VirtualDatabase.

From the application's viewpoint, VirtualDatabase refers to a set of specific database access information, which can achieve governance capability by binding corresponding governance strategies.

2. DatabaseEndpoint

In database governance, VirtualDatabase declares the logical database available to applications, but actually, data storage depends on a physical database. Here it is called database access endpoint, namely DatabaseEndpoint.

DatabaseEndpoint is imperceptible to applications. It can only be bound to VirtualDatabase by a specific governance strategy and then be connected and used.

3. ReadWriteSplitting

Read/write splitting is a commonly used database extension method. The primary database is used for transactional read/write operations, while the secondary database is mainly used for queries.

4. Sharding

Data sharding is an extension strategy based on data attributes. Once data attributes are calculated, requests are sent to a specific data backend. Currently, sharding consists of sharding with shard keys and automatic sharding. Sharding with shard keys needs to specify which table or column is to be sharded and the algorithms used for sharding.

5. Encryption

To meet the requirements of audit security and compliance, enterprises need to provide strict security measures for data storage, such as data encryption.

Data encryption parses the SQL entered by users and rewrites SQL according to the encryption rules provided by users.

By doing so, the plaintext data can be encrypted, and plaintext data (optional) and ciphertext data can be both stored in the underlying database.

When the user queries the data, Encryption only takes the ciphertext data from the database, decrypts it, and finally returns the decrypted original data to the user.

6. Shadow

Shadow database can receive grayscale traffic or data test requests in a grayscale environment or test environment, and flexibly configure multiple routing methods combined with shadow algorithms.

7. DatabaseDiscovery

Database auto-discovery can detect the change of data source status according to the high availability configuration of the database and then make adjustments to traffic strategy accordingly.

For example, if the backend data source is MySQL MGR, you can set the database discovery type as MYSQL.MGR, specify group-name and configure corresponding heartbeat intervals.

8. DistributedTransaction

It can declare configurations related to distributed transactions. Users can declare the transaction types without additional configuration.

Database Governance Example

# Virtual database configuration
apiVersion: database.opensergo.io/v1alpha1
kind: VirtualDatabase
metadata:
  name: sharding_db
spec:
  services:
  - name: sharding_db
    databaseMySQL:
      db: sharding_db
      host: localhost
      port: 3306
      user: root
      password: root
    sharding: "sharding_db"  # Declare the desired sharding strategy.
---
# The database endpoint configuration of the first data source
apiVersion: database.opensergo.io/v1alpha1
kind: DatabaseEndpoint
metadata:
  name: ds_0
spec:
  database:
    MySQL:                 # Declare the backend data source type and other related information.
      url: jdbc:mysql://192.168.1.110:3306/demo_ds_0?serverTimezone=UTC&useSSL=false
      username: root
      password: root
      connectionTimeout: 30000
      idleTimeoutMilliseconds: 60000
      maxLifetimeMilliseconds: 1800000
      maxPoolSize: 50
      minPoolSize: 1      
---
# The database endpoint configuration of the second data source
apiVersion: database.opensergo.io/v1alpha1
kind: DatabaseEndpoint
metadata:
  name: ds_1
spec:
  database:
    MySQL:                              # Declare the backend data source type and other related information.
      url: jdbc:mysql://192.168.1.110:3306/demo_ds_1?serverTimezone=UTC&useSSL=false
      username: root
      password: root
      connectionTimeout: 30000
      idleTimeoutMilliseconds: 60000
      maxLifetimeMilliseconds: 1800000
      maxPoolSize: 50
      minPoolSize: 1
---
# Sharding configuration
apiVersion: database.opensergo.io/v1alpha1
kind: Sharding
metadata:
  name: sharding_db
spec:
  tables: # map[string]object type
    t_order:
      actualDataNodes: "ds_${0..1}.t_order_${0..1}"
      tableStrategy:
        standard:
          shardingColumn: "order_id"
          shardingAlgorithmName: "t_order_inline"
      keyGenerateStrategy:
        column: "order_id"
        keyGeneratorName: "snowflake"
    t_order_item:
      actualDataNodes: "ds_${0..1}.t_order_item_${0..1}"
      tableStrategy:
        standard:
          shardingColumn: "order_id"
          shardingAlgorithmName: "t_order_item_inline"
      keyGenerateStrategy:
        column: order_item_id
        keyGeneratorName: snowflake
  bindingTables:
  - "t_order,t_order_item"
  defaultDatabaseStrategy:
    standard:
     shardingColumn: "user_id"
     shardingAlgorithmName: "database_inline"
  # defaultTableStrategy: # Null means none 
  shardingAlgorithms: # map[string]object type
    database_inline:
      type: INLINE    
      props: # map[string]string type
        algorithm-expression: "ds_${user_id % 2}"
    t_order_inline:  
      type: INLINE    
      props:
        algorithm-expression: "d_order_${order_id % 2}"      
    t_order_item_inline:
      type: INLINE    
      props:
        algorithm-expression: "d_order_item_${order_id % 2}"
  keyGenerators: # map[string]object type
    snowflake:
      type: SNOWFLAKE

About Apache ShardingSphere

Apache ShardingSphere is a distributed database ecosystem that can transform any database into a distributed database and enhance it with sharding, elastic scaling, encryption features & more.

🔗 Apache ShardingSphere Useful Links

About OpenSergo

OpenSergo is an open and universal service governance specification that is oriented toward distributed service architecture and covers a full-link heterogeneous ecosystem.

It is formed based on the industry's service governance scenarios and practices. The biggest characteristic of OpenSergo is defining service governance rules with a unified set of configuration/DSL/protocol and is oriented towards multi-language heterogeneous architecture, achieving full-link ecosystem coverage.

No matter if the microservice language is Java, Go, Node.js, or some other language, or whether it’s a standard microservice or Mesh-based access, developers can use the same set of OpenSergo CRD standard configurations.

This allows developers to implement unified governance and control for each layer, ranging from the gateway to microservices, from database to cache, and from registration and discovery to the configuration of services.

🔗 OpenSergo GitHub

ElasticJob 3.0.2 is released including failover optimization, scheduling stability, and Java 19 compatibility

Apache ShardingSphere — Mon, 31 Oct 2022 05:51:47 +0000

ElasticJob, one of the sub-projects of the Apache ShardingSphere community, is a distributed scheduling solution oriented towards Internet applications and massive tasks.

Since ElasticJob 3.0.1 was released, we’ve received a lot of feedback from users on GitHub. After merging the code committed by the contributors, ElasticJob has been significantly optimized.

The resulting ElasticJob 3.0.2 has been improved in terms of Failover, scheduling stability, job API, and Java 19 compatibility.

Release 3.0.2 Key Features

Failover Optimization

Failover is an important ElasticJob feature.

In ElasticJob 3.0.1, the Failover feature had some limitations and disadvantages:

Failover could take effect with at least two job instances. Assume that there’s only one job instance executing tasks and a fault occurs, Failover cannot take effect even if the instance recovers or a new job instance is started after the fault occurs. In this case, the old Failover feature required redundant resources. Particularly, if sharding was not needed for the job, the backup instance would be idle for a long time and continuously occupy resources.
Failover was sensitive. Even if the job was not in execution, as long as one instance goes offline, failover of another instance would still be triggered to immediately execute the sharding owned by the offline instance. In other words, failover was triggered by the offline instance whether it is a normal offline or a fault occurs, which is disturbing for users.

ElasticJob 3.0.2 optimized the Failover feature.

Suppose a single job instance is executing the tasks and it exits due to a fault in the execution process. After a fault occurs, a new job instance is started, which can detect the failed sharding and trigger Failover immediately. For example, we deploy a job instance with one copy in the Kubernetes environment. If the job instance exits unexpectedly due to a fault, Kubernetes automatically starts a new job instance. By this time, Failover will be triggered and continue to execute the previously failed job.
The optimized Failover takes effect only for ongoing jobs. If no job is being executed when a fault occurs, Failover is not triggered. In this case, Failover will not be triggered when the job instance goes offline normally.

Separating the event notification threads of different jobs to avoid an endless loop.

ElasticJob achieves distributed coordination through ZooKeeper. In practical scenarios, users may start multiple jobs in the same project simultaneously, all of which use the same Apache Curator client. There are certain risks due to the nature of ZooKeeper and the callback method of Curator in a single event thread.

Callbacks will increase accordingly if a huge number of jobs are triggered frequently. The processing capability of Curator event threads may reach an upper limit, resulting in delayed triggering or failure to trigger.
If one job blocks the event thread accidentally, other jobs will also be affected, leading to delayed triggering or failure to trigger.

In ElasticJob 3.0.2, the callback method of each job will be executed on the thread to which the job belongs, preventing jobs from affecting each other.

Moreover, ElasticJob 3.0.2 modified the code which may lead to an endless loop. When a job instance is online and the server nodes in ZooKeeper change (such as being deleted), ElasticJob is stuck in determining whether the server is enabled. This problem has been tackled.

Java 19 Support

Java 19 was released on September 20, 2022. ElasticJob’s code closely follows Java versions and currently can be built and used through Java 19. Elasticjob now supports Java 8 through 19.

Release Notes

Bug Fixes

Fix itemErrorMessages not cleared after the job finished.
Fix Curator notify thread may be blocked and avoid probably endless loop in ServerService.
Fix the problem that NPE may occur in the deserialization of job instance ID and job configuration.
Fix failover to sensitive.

Enhancements

Script Job exception’s stack was ignored.
Support using different event trace data sources when using Spring Boot.
Supports building projects with Java 19.

Please refer to GitHub ElasticJob Milestone for details.

Useful Links

🔗 ElasticJob Milestone

🔗 Release Notes

🔗 ShardingSphere ElasticJob GitHub Address

🔗 ShardingSphere Project Address

🔗 ShardingSphere ElasticJob Official Website

Author

Wu Weijie, is an infrastructure R&D engineer at SphereEx, and an Apache ShardingSphere PMC.

He now focuses on the R&D of the Apache ShardingSphere access port and the sub-project ElasticJob.

ShardingSphere 5.2.1 is released — Here are the highlights

Apache ShardingSphere — Thu, 27 Oct 2022 06:15:25 +0000

Our new 5.2.1 release brings new features and enhancements such as the ShardingSphere system database, SQL HINT mandatory sharding routing, asynchronous data consistency check, and support for Consul and Nacos governance center.

Introduction

Following 1.5 months in development, Apache
ShardingSphere 5.2.1 is released. Our community merged 614 PRs from teams and individuals around the world. The resulting 5.2.1 release has been optimized in terms of features, performance, testing, documentation, examples, etc.

The new release brings the following highlights:

ShardingSphere system database
SQL HINT mandatory sharding routing
Asynchronous data consistency check

Version 5.2.1 introduces a new ShardingSphere system database to provide statistical information for distributed databases. The statistical information can help the SQL Federation execution engine evaluate the execution costs to select an execution plan with the lowest costs. Moreover, ShardingSphere can collect the distribution information after data sharding, providing references for automatic sharding scaling management.

SQL HINT is an important feature provided by ShardingSphere 5. x, with which users can flexibly control routes through SQL annotations. The new release enhances the SQL HINT capability and supports data-sharding mandatory routing.

Users only need to add annotations in the format: /* SHARDINGSPHERE_HINT: t_order.SHARDING_DATABASE_VALUE=1, t_order.SHARDING_TABLE_VALUE=1 */ before transferring sharding values to the ShardingSphere routing engine. Compared with the former HintManager method, SQL HINT is more flexible without requiring users to write code.

The new version also enhances the data consistency check capability during data migration, enabling asynchronous execution of the check. Users can view the data migration progress through DistSQL, which improves the ease of data migration.

Moreover, the governance center now supports Consul and Nacos, providing more choices for users. SQL compatibility is greatly improved. Read/write splitting has also been optimized, as Proxy can be started forcibly by using the -f parameter after the secondary database is disabled.

Next, this post will introduce the updates of ShardingSphere 5.2.1 in detail.

Notable Highlights

ShardingSphere System Database Like MySQL and PostgreSQL system databases, ShardingSphere 5.2.1 introduces a new system database to manage the metadata of databases.

ShardingSphere system database mainly stores dynamic and static metadata. Dynamic metadata refers to frequently updated data, such as statistical information from distributed databases, which needs to be collected and maintained regularly through built-in scheduling tasks.

In comparison, static metadata refers to data that will remain the same without user modification. An example is the status parameter of the ShardingSphere distributed database set by users, which only needs to be stored in the metadata database for querying.

As this function is currently in the experimental phase, you need to start collecting metadata through the following configuration.

proxy-metadata-collector-enabled: true

Connect Proxy through the command line and then execute the SHOW DATABASES statement. We can see the new shardingsphere system database which stores the metadata of distributed databases.

Currently, the sharding_table_statistics table is added to the shardingsphere system database, used for collecting the distribution information of sharding tables, including row_count and size.

Statistical information provided by the ShardingSphere system database can help the SQL Federation execution engine evaluate the execution cost. This allows for the selection of a suitable association order and method and achieves efficient execution.

Moreover, by collecting the data distribution information and loading information of storage nodes, ShardingSphere can carry out automatic sharding scaling, reducing the operation & maintenance costs for users.

SQL HINT Mandatory Sharding Routing

In some special business scenarios, the fields used for sharding exist in external business logic rather than SQL, database, and table structures. Currently, Hint is needed to introduce sharding key value to complete sharding routing.

Before version 5.2.1, there were two ways to use Hint. One way is to use it through HintManager in the JDBC access port, while another way is to start the proxy-hint-enabled parameter in the Proxy access port.

In the first case, users need to write codes to call the addDatabaseShardingValue and addTableShardingValue methods to set the values for the database and table sharding. However, if the HintManager method is used, users have to modify the original logic at some cost.

// Sharding database and table with using HintManagerString sql = "SELECT * FROM t_order";
try (HintManager hintManager = HintManager.getInstance();
     Connection conn = dataSource.getConnection();
     PreparedStatement preparedStatement = conn.prepareStatement(sql)) {
    hintManager.addDatabaseShardingValue("t_order", 1);
    hintManager.addTableShardingValue("t_order", 2);
    try (ResultSet rs = preparedStatement.executeQuery()) {
        while (rs.next()) {
            // ...        }
    }
}

If it’s the second case, start the proxy-hint-enabled parameter first, and then clear the sharding values through the following DistSQL configuration.

Nevertheless, the Hint-based method will transform Proxy's thread processing model from IO multiplexing to one independent thread for each request, which will reduce Proxy's throughput. This means that users have to make tradeoffs.

-- For the current connection, add sharding values yy to table xx. xx: logical table name; yy: database sharding value
ADD SHARDING HINT DATABASE_VALUE t_order= 100;
-- For the current connection, add sharding values yy to table xx. xx: logical table name; yy: table sharding value
ADD SHARDING HINT TABLE_VALUE t_order = 100;
-- For the current connection, clear sharding hint settings
CLEAR SHARDING HINT;

As there are problems with both methods, version 5.2.1 adds the SQL Hint mandatory sharding routing feature. This allows users to control routes flexibly through SQL annotations. There’s no need to modify the logic of the original code and the thread processing model in the Proxy access port is not affected.

Before using the SQL Hint mandatory sharding routing, users should enable configurations of parsing annotations in advance and set sqlCommentParseEnabled to true. The annotation format only supports /* */ and content has to start with SHARDINGSPHERE_HINT:.

Optional properties include:

{table}.SHARDING_DATABASE_VALUE: Used to add data source sharding key value corresponding to {table}. Multiple properties are separated by a comma.
{table}.SHARDING_TABLE_VALUE: Used to add table sharding key value corresponding to {table}. Multiple properties are separated by a comma.

When database sharding is required and table sharding is not, and mandatory routing is carried out to a sub-database, you can use SHARDING_DATABASE_VALUE to add shards without specifying {table}.

The following is an example of using the HINT_INLINE algorithm.

rules:
- !SHARDING
  tables:
    t_order:
      actualDataNodes: ds_${0..1}.t_order_${0..1}
      databaseStrategy:
        hint:
          shardingAlgorithmName: database_hint_inline
      tableStrategy:
        hint:
          shardingAlgorithmName: t_order_hint_inline
  defaultDatabaseStrategy:
    none:
  defaultTableStrategy:
    none:
  shardingAlgorithms:
    database_hint_inline:
      type: HINT_INLINE
      props:
        algorithm-expression: ds_${value % 2}
    t_order_hint_inline:
      type: HINT_INLINE
      props:
        algorithm-expression: t_order_${value % 2}

We can transfer the sharding value of the t_order table through SQL Hint. We can see from the PREVIEW statement that although SQL statements have no specified sharding key, the sharding key-value pair introduced by SQL Hint can also achieve sharding routing.

/* SHARDINGSPHERE_HINT: t_order.SHARDING_DATABASE_VALUE=1, t_order.SHARDING_TABLE_VALUE=1 */
SELECT * FROM t_order;
PREVIEW /* SHARDINGSPHERE_HINT: t_order.SHARDING_DATABASE_VALUE=1, t_order.SHARDING_TABLE_VALUE=1 */ SELECT * FROM t_order;
+------------------+----------------------------------------------------------------------------------------------------------------------+
| data_source_name | actual_sql                                                                                                           |
+------------------+----------------------------------------------------------------------------------------------------------------------+
| ds_1             | /* SHARDINGSPHERE_HINT: t_order.SHARDING_DATABASE_VALUE=1, t_order.SHARDING_TABLE_VALUE=1 */ SELECT * FROM t_order_1 |
+------------------+----------------------------------------------------------------------------------------------------------------------+

Asynchronous data consistency check

Before version 5.2.1, users had to wait for the server to return check results synchronously. Sometimes the check was time-consuming, so timeout could occur in the session of database servers.

In such cases, users couldn’t observe the result and could only check logs, which were not user-friendly. In response to the problem, version 5.2.1 supports asynchronous data consistency check capability as well as a set of DistSQL as follows:

CHECK MIGRATION jobId — Asynchronous data consistency check
SHOW MIGRATION CHECK STATUS jobId — Query the check progress
START MIGRATION CHECK jobId — Start the check job
STOP MIGRATION CHECK jobId — Stop the check job

Through the above DistSQL, users can manage the data migration more easily. Examples are as follows:

-- Execute consistency check
mysql> CHECK MIGRATION 'j0101270cbb6714cff0d7d13db9fa23c7c0a1' BY TYPE (NAME='DATA_MATCH');
Query OK, 0 rows affected (0.24 sec)
-- Query the check progress
mysql> SHOW MIGRATION CHECK STATUS 'j0101270cbb6714cff0d7d13db9fa23c7c0a1';
+---------+--------+---------------------+-------------------+-------------------------+----------------+------------------+---------------+
| tables  | result | finished_percentage | remaining_seconds | check_begin_time        | check_end_time | duration_seconds | error_message |
+---------+--------+---------------------+-------------------+-------------------------+----------------+------------------+---------------+
| t_order | false  | 0                   | 2450              | 2022-10-12 16:07:17.082 |                | 14               |               |
+---------+--------+---------------------+-------------------+-------------------------+----------------+------------------+---------------+
1 row in set (0.02 sec)
-- Stop the check job
mysql> STOP MIGRATION CHECK 'j0101270cbb6714cff0d7d13db9fa23c7c0a1';
Query OK, 0 rows affected (0.06 sec)
-- Start the check job
mysql> START MIGRATION CHECK 'j0101270cbb6714cff0d7d13db9fa23c7c0a1';
Query OK, 0 rows affected (5.13 sec)

Please refer to the official documentation and stay tuned for future posts with more detailed information.

Enhancements

Kernel

This update optimized the SQL parsing of Oracle and PostgreSQL, greatly improving the SQL compatibility of ShardingSphere. Detailed SQL parsing optimization can be found in the release notes section below.

It’s a long-term mission for the ShardingSphere community to improve SQL parsing compatibility, and anyone interested is welcome to join us.

Version 5.2.1 also supports read/write splitting. After the secondary database is disabled, ShardingSphere-Proxy can be started forcibly by using the -f parameter.

In this case, the Proxy can still be started for ops & maintenance work even if part of the storage nodes are unavailable. Moreover, the kernel function has been optimized in terms of the display results of the SHOW PROCESSLIST statement and added thread display of the Sleep status.

By optimizing the logic of unicast routing and reusing cached database connections as much as possible, ShardingSphere's execution performance is improved.

Elastic scaling

The elastic scaling module only supports data migration for tables with a unique key, and tables are allowed to migrate to new table names. Moreover, it optimized the use of thread pools in data migration. The data consistency check can be interrupted, which makes it easier for users to manage data migration.

Distributed Governance

In terms of distributed governance, the governance center is now able to support Consul and Nacos, providing you with more choices. At the same time, the built-in H2 database is used to support persisting metadata information in a standalone pattern.

In the running mode, the overwrite configuration item is removed. Metadata information is based on the data in the governance center, and users can dynamically change the rules and configurations through DistSQL.

Release Notes

In the following sections, you will find the release notes of ShardingSphere 5.2.1. To improve user experience, the API of some functions is modified in this update. Please refer to the API changes section below for details.

New Features

Add ShardingSphere default system database to support global metadata management
Support asynchronous data consistency check
Added support for Consul governance center
Added support for Nacos governance center
Added support for the view function in the governance center

Enhancements

SQL Federation engine adds ADVANCED executor and adapts to openGauss database
Support ShardingSphere Proxy startup after read-write splitting read database is disabled
SQL HINT supports force sharding route
Show processlist supports showing Proxy connections (sleep, active)
Optimized ShardingSphere-JDBC data source configuration error message
ShardingSphere-JDBC supports SpringBoot 3. x version
Support load MySQL, PostgreSQL, openGauss, and SQLServer view metadata
Update Snakeyaml to 1.33 and open YAML 3MB limit
Reuse cached connections as possible when unicast sharding
Support Parsing ALTER ROLE in Oracle
Add support of ALTER RESOURCE COST for Oracle
Support parsing Drop Materialized View in Oracle
Support parsing DROP LIBRARY in Oracle
Support parsing DROP JAVA in Oracle
Support parsing DROP PLUGGABLE DATABASE in Oracle
Support parsing DROP INDEX TYPE in Oracle
Support Parsing ALTER PACKAGE in openGauss
Support openGauss select offset, count statement parse, and remove useless syntax in PostgreSQL grammar
Add max_size to openGauss syntax
Optimize alter view/drop view parse logic and fix alter view refresher bug
Add SQL parser error detail to ParseCancellationException
Add support for parse OptOnConflict for Postgres
Enhance support for ALTER TABLE and ALTER VIEW in PostgreSQL
Add a missing keyword in the PostgreSQL Declare Statement
Add JSON function support to MySQL parser
ShardingSphere-Proxy automatically adapts to cgroup memory limits in the Docker environment
Show migration status DistSQL respond new added error_message column
Show migration status respond new added processed_records_count column
Support MySQL 8 caching_sha2_password authentication in incremental dump
Improve drop pipeline process configuration
Support unique key table migration
Support migrating table to new name table
Improve thread pool usage in pipeline task and runner
Support cancelable data consistency check
DistSQL: When creating or altering the readwrite-splitting rule, check duplicate write or read resources
DistSQL: Add validity check for ALTER SHARDING BINDING TABLE RULES
Standalone mode H2 supports persistent metadata
Fix openGauss cursor execution in xa transaction
Added transaction-related exceptionsBug Fix

Bug Fixes

Generate proper placeholder for PostgreSQL when rewriting
Fix opengauss update set parse exception
Fix parse exception when executing insert statement with a negative value
Fix wrong connectDescriptorUrlPattern in OracleDataSourceMetaData
Fix insert SQL garbled after sharding rewrote in special rules
Fix exception when execute select * from information_schema.tables
Fix exception when executing alter view rename
Fix PostgreSQL check data source permission when using rolsuper
DistSQL: fix NPE for REFRESH TABLE METADATA when there is no resource
Fix Unmodified table metadata when modifying rules
Fix database discovery
The MySQL.NORMAL_REPLICATION algorithm cannot find the primary node”
Fixed using etcd to build a cluster event not aware
Fix NPE occurred when transaction management is not created
Fix routing exception that occurs when the sharding value of the InlineShardingAlgorithm algorithm exceeds Integer

API Changes

SQL HINT syntax format adjust to SQL-style format
DistSQL: Remove syntax COUNT DATABASE RULES
ShardingSphere mode removes overwrite configuration
Agent: Optimize configuration of agent.yaml

Relevant Links

🔗 Download Link

🔗 Release Notes

🔗 Project Address

🔗 Cloud Sub-project Address

Community Contribution

This Apache ShardingSphere 5.2.1 release is the result of 614 merged PRs, committed by 38 contributors. Thank you for your efforts!

ShardingSphere 5.2.0: Audit for sharding intercepts unreasonable requests in multi-shards scenarios

Apache ShardingSphere — Mon, 17 Oct 2022 06:55:38 +0000

1. Background

Thanks to our continuous review of the ShardingSphere’s community feedback that we use to develop features such as data sharding and read/write splitting, we found that some users create a large number of shards when using the data sharding feature.

In such cases, there can be 1,000 physical tables corresponding to a sharding logical table, which largely disturbs users.

For instance, a SELECT * FROM t_order statement will lead to a full-route, which is obviously not the case for OLTP. This SQL can be placed in another Proxy to avoid blocking other requests.

However, if users are not familiar with Proxy, or write a where condition and don't know that sharding is not supported in this condition, a full-route is still required.

A full route can lower the performance of Proxy and even result in the failure of a reasonable request. Imagine that there are 1000 shards in a physical database, if they are executed in parallel, 1,000 connections are needed — and if in serial, the request can lead to a timeout. In this regard, community users requested whether the unreasonable request can be intercepted directly.

We’ve considered the issue for a while. If we simply block the full-route operation, we just need to check it in the code and add a switch to the configuration file. On the other hand, if the user later needs to set a table to read-only or requires the update operation to carry a limit, does that mean we need to change the code and configuration again? This obviously goes against the pluggable logic of Proxy.

In response to the above problems, the recently released Apache ShardingSphere 5.2.0 provides users with SQL audit for the sharding function. The audit can either be an interception operation or a statistical operation. Similar to the sharding and unique key generation algorithms, the audit algorithm is also plugin-oriented, user-defined, and configurable.

Next, we will elaborate on the implementation logic for data sharding’s audit, with specific SQL examples.

2. Audit for sharding interface

The entrance to Apache ShardingSphere’s audit is in the org.apache.shardingsphere.infra.executor.check.SQLCheckEngine class, which will invoke the check method of the SQLChecker interface. Currently, ShardingSphere audit contains audit for permission (verify username and password) and audit for sharding.

Here we focus on the parent interface implemented in ShardingAuditChecker of audit for sharding.

We can learn its working principles quickly through viewing the check code of org.apache.shardingsphere.sharding.checker.audit.ShardingAuditChecker.

public interface ShardingAuditAlgorithm extends ShardingSphereAlgorithm {

    /**
     * Sharding audit algorithm SQL check.
     *
     * @param sqlStatementContext SQL statement context
     * @param parameters SQL parameters
     * @param grantee grantee
     * @param database database
     * @return SQL check result
     */
    SQLCheckResult check(SQLStatementContext<?> sqlStatementContext, List<Object> parameters, Grantee grantee, ShardingSphereDatabase database);
}

This method obtains the audit strategies of all the sharding tables involved and invokes the audit algorithms configured in each sharding table audit strategy. If an audit algorithm fails to pass, an exception is displayed to the user.

Some users may wonder what disableAuditNames does here. The audit for sharding also allows users to skip this process. In some cases, users may need to execute SQL that should have been blocked by the audit, and they are aware of the impact of this SQL.

For this reason, we provide Hint: disableAuditNames to skip audit interception, which will be described with practical examples later on. The Proxy Administrators can configure allowHintDisable to control whether to allow users to skip this process. The default value is true, indicating that Hint-based skip is allowed.

3. Audit for sharding algorithm

The audit for sharding algorithm interface org.apache.shardingsphere.sharding.spi.ShardingAuditAlgorithm is inherited from SPI class ShardingSphereAlgorithm. It inherits type and props properties and defines its own check method. If you‘re looking to customize your own audit algorithm, just implement the interface and add it to INF.services.

public interface ShardingAuditAlgorithm extends ShardingSphereAlgorithm {

    /**
     * Sharding audit algorithm SQL check.
     *
     * @param sqlStatementContext SQL statement context
     * @param parameters SQL parameters
     * @param grantee grantee
     * @param database database
     * @return SQL check result
     */
    SQLCheckResult check(SQLStatementContext<?> sqlStatementContext, List<Object> parameters, Grantee grantee, ShardingSphereDatabase database);
}

Apache ShardingSphere implements a general audit for sharding algorithm org.apache.shardingsphere.sharding.algorithm.audit.DMLShardingConditionsShardingAuditAlgorithm, namely the above-mentioned SQL statement that intercepts the full-route.

The algorithm makes decisions by determining whether the sharding condition is null. Of course, it won't intercept broadcast tables and non-sharding tables.

public final class DMLShardingConditionsShardingAuditAlgorithm implements ShardingAuditAlgorithm {

    @Getter
    private Properties props;

    @Override
    public void init(final Properties props) {
        this.props = props;
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    @Override
    public SQLCheckResult check(final SQLStatementContext<?> sqlStatementContext, final List<Object> parameters, final Grantee grantee, final ShardingSphereDatabase database) {
        if (sqlStatementContext.getSqlStatement() instanceof DMLStatement) {
            ShardingRule rule = database.getRuleMetaData().getSingleRule(ShardingRule.class);
            if (rule.isAllBroadcastTables(sqlStatementContext.getTablesContext().getTableNames())
                    || sqlStatementContext.getTablesContext().getTableNames().stream().noneMatch(rule::isShardingTable)) {
                return new SQLCheckResult(true, "");
            }
            ShardingConditionEngine shardingConditionEngine = ShardingConditionEngineFactory.createShardingConditionEngine(sqlStatementContext, database, rule);
            if (shardingConditionEngine.createShardingConditions(sqlStatementContext, parameters).isEmpty()) {
                return new SQLCheckResult(false, "Not allow DML operation without sharding conditions");
            }
        }
        return new SQLCheckResult(true, "");
    }

    @Override
    public String getType() {
        return "DML_SHARDING_CONDITIONS";
    }
}

Here we’d like to introduce another audit for sharding algorithm: LimitRequiredShardingAuditAlgorithm. This algorithm can intercept SQL without carrying limit in the update and delete operations.

As this algorithm is less universal, it is not currently integrated into Apache ShardingSphere. As you can see, it is very easy to implement a custom algorithm, which is why we need to design the audit for sharding framework. Thanks to its plugin-oriented architecture, ShardingSphere boasts great scalability.

public final class LimitRequiredShardingAuditAlgorithm implements ShardingAuditAlgorithm {

    @Getter
    private Properties props;

    @Override
    public void init(final Properties props) {
        this.props = props;
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    @Override
    public SQLCheckResult check(final SQLStatementContext<?> sqlStatementContext, final List<Object> parameters, final Grantee grantee, final ShardingSphereDatabase database) {
        if (sqlStatementContext instanceof UpdateStatementContext && !((MySQLUpdateStatement) sqlStatementContext.getSqlStatement()).getLimit().isPresent()) {
            return new SQLCheckResult(false, "Not allow update without limit");
        }
        if (sqlStatementContext instanceof DeleteStatementContext && !((MySQLDeleteStatement) sqlStatementContext.getSqlStatement()).getLimit().isPresent()) {
            return new SQLCheckResult(false, "Not allow delete without limit");
        }
        return new SQLCheckResult(true, "");
    }

    @Override
    public String getType() {
        return "LIMIT_REQUIRED";
    }
}

4. Using audit for sharding

Audit for sharding requires you to configure audit strategy for logical tables. To help you quickly get started, its configuration is the same with that of the sharding algorithm and the sharding key value generator.

There is an algorithm definition and strategy definition, and default audit strategy is also supported. If the audit strategy is configured in the logical table, it takes effect only for the logical table.

If defaultAuditStrategy is configured in the logical table, it takes effect fo all the logical tables under the sharding rule. Auditors are similar to ShardingAlgorithms, auditStrategy to databaseStrategy, and defaultAuditStrategy to defaultDatabaseStrategy or defaultTableStrategy.

Please refer to the following configuration. Only the configuration of audit for sharding is displayed. You need to configure the sharding algorithm and data source by yourself.

rules:
  - !SHARDING
    tables:
      t_order:
        actualDataNodes: ds_${0..1}.t_order_${0..1}
        auditStrategy:
          auditorNames:
            - sharding_key_required_auditor
          allowHintDisable: true
    defaultAuditStrategy:
      auditorNames:
        - sharding_key_required_auditor
      allowHintDisable: true
    auditors:
      sharding_key_required_auditor:
        type: DML_SHARDING_CONDITIONS

Step 1: Execute a query operation. An error is displayed as the audit strategy for intercepting the full-database route is configured.

mysql> select * from t_order;
ERROR 13000 (44000): SQL check failed, error message: Not allow DML operation without sharding conditions

Step 2: Add HINT. The name of the HINT is /* ShardingSphere hint: disableAuditNames */，and disableAuditNames is followed by the auditorsNames configured in the preceding command.

If there are multiple names, separate them with spaces such as /* ShardingSphere hint: disableAuditNames=auditName1 auditName2*/. After using HINT, we can see that the SQL operation is successfully executed.

mysql> /* ShardingSphere hint: disableAuditNames=sharding_key_required_auditor */ select * from t_order;
+----------+---------+------------+--------+
| order_id | user_id | address_id | status |
+----------+---------+------------+--------+
|       30 |      20 |         10 | 20     |
|       32 |      22 |         10 | 20     |
+----------+---------+------------+--------+
2 rows in set (0.01 sec)

Note: HINT requires you to modify the server.yaml configuration of Proxy. In addition, if you are using MySQL terminal to connect to Proxy directly, you need to add the -c property — otherwise, HINT comments will be filtered out of the MySQL terminal and will not be parsed by Proxy on the backend.

rules:
  - !SQL_PARSER
    sqlCommentParseEnabled: true
    sqlStatementCache:
      initialCapacity: 2000
      maximumSize: 65535
    parseTreeCache:
      initialCapacity: 128
      maximumSize: 1024
props:
  proxy-hint-enabled: true
mysql -uroot -proot -h127.0.0.1 -P3307  -c

5. DistSQL with audit for sharding

Currently, as you can see from the release notes Apache ShardingSphere 5.2.0 supports the following DistSQL with audit for sharding function.

CREATE SHARDING AUDITOR
ALTER SHARDING AUDITOR
SHOW SHARDING AUDIT ALGORITHMS

The following DistSQL will be supported in future releases:

DROP SHARDING AUDITOR
SHOW UNUSED SHARDING AUDIT ALGORITHMS
CREATE SHARDING TABLE RULE # including AUDIT_STRATEGY

This post introduced how audit for sharding works with specific examples. I believe you already have basic understanding of this function, and you can use it whenever you need or use custom algorithm.

You are also welcome to submit general algorithms to the community. If you have any ideas you’d like to contribute or you encounter any issues with your ShardingSphere, feel free to post them on Github.

Author

Huang Ting, a technology engineer at Tencent Financial Technology (FiT) & ShardingSphere Committer.

He is mainly responsible for the R&D of Proxy-related audit for sharding and transaction features.

How does ShardingSphere’s Show processlist & Kill Work?

Apache ShardingSphere — Thu, 22 Sep 2022 10:30:31 +0000

For those of you who often use databases, you may wonder:

How do I check what SQL is currently being executed by the database, and in what state?
How do I terminate abnormal SQL? For instance, if a SELECT statement used to query a table with massive data does not carry query conditions, it would drag down the performance of the entire database. This may push to want to terminate this abnormal SQL.

In response to the above issues, Apache ShardingSphere introduced functions such as Show processlist and Kill <processID>.

1. Introduction

Show processlist: this command can display the list of SQL currently being executed by ShardingSphere and the execution progress of each SQL. If ShardingSphere is deployed in cluster mode, the Show processlist function aggregates the SQL running for all Proxy instances in the cluster and then displays the result, so you can always see all the SQL running at that moment.

mysql> show processlist \G;
*************************** 1. row ***************************
     Id: 82a67f254959e0a0807a00f3cd695d87
   User: root
   Host: 10.200.79.156
     db: root
Command: Execute
   Time: 19
  State: Executing 0/1
   Info: update t_order set version = 456
1 row in set (0.24 sec)

Kill <processID>: This command is implemented based on Show processlist and can terminate the running SQL listed in the Show processlist.

mysql> kill 82a67f254959e0a0807a00f3cd695d87;
Query OK, 0 rows affected (0.17 sec)

2. How do they work?

Now that you understand the functions of Show processlist and Kill <processID>, let's see how the two commands work. As the working principle behind Kill <processID> is similar to that of Show processlist, we'll focus on the interpretation of Show processlist.

2.1 How is SQL saved and destroyed?

Each SQL executed in ShardingSphere will generate an ExecutionGroupContext object. The object contains all the information about this SQL, among which there is an executionID field to ensure its uniqueness.

When ShardingSphere receives a SQL command, the GovernanceExecuteProcessReporter# report is called to store ExecutionGroupContext information into the cache of ConcurrentHashMap (currently only DML and DDL statements of MySQL are supported; other types of databases will be supported in later versions. Query statements are also classified into DML).

public final class GovernanceExecuteProcessReporter implements ExecuteProcessReporter {

    @Override
    public void report(final QueryContext queryContext, final ExecutionGroupContext<? extends SQLExecutionUnit> executionGroupContext,
                       final ExecuteProcessConstants constants, final EventBusContext eventBusContext) {
        ExecuteProcessContext executeProcessContext = new ExecuteProcessContext(queryContext.getSql(), executionGroupContext, constants);
        ShowProcessListManager.getInstance().putProcessContext(executeProcessContext.getExecutionID(), executeProcessContext);
        ShowProcessListManager.getInstance().putProcessStatement(executeProcessContext.getExecutionID(), executeProcessContext.getProcessStatements());
    }
}
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class ShowProcessListManager {

    private static final ShowProcessListManager INSTANCE = new ShowProcessListManager();

    @Getter
    private final Map<String, ExecuteProcessContext> processContexts = new ConcurrentHashMap<>();

    @Getter
    private final Map<String, Collection<Statement>> processStatements = new ConcurrentHashMap<>();

    public static ShowProcessListManager getInstance() {
        return INSTANCE;
    }

    public void putProcessContext(final String executionId, final ExecuteProcessContext processContext) {
        processContexts.put(executionId, processContext);
    }

    public void putProcessStatement(final String executionId, final Collection<Statement> statements) {
        if (statements.isEmpty()) {
            return;
        }
        processStatements.put(executionId, statements);
    }
}

As shown above, the ShowProcessListManager class has two cache Maps, namely processContexts and processStatements. The former stores the mapping between executionID and ExecuteProcessContext.

The latter contains the mapping between executionID and Statement objects that may generate multiple statements after the SQL is overwritten.

Every time ShardingSphere receives a SQL statement, the SQL information will be cached into the two Maps. After SQL is executed, the cache of Map will be deleted.

@RequiredArgsConstructor
public final class ProxyJDBCExecutor {

    private final String type;

    private final ConnectionSession connectionSession;

    private final JDBCDatabaseCommunicationEngine databaseCommunicationEngine;

    private final JDBCExecutor jdbcExecutor;

    public List<ExecuteResult> execute(final QueryContext queryContext, final ExecutionGroupContext<JDBCExecutionUnit> executionGroupContext,
                                       final boolean isReturnGeneratedKeys, final boolean isExceptionThrown) throws SQLException {
        try {
            MetaDataContexts metaDataContexts = ProxyContext.getInstance().getContextManager().getMetaDataContexts();
            EventBusContext eventBusContext = ProxyContext.getInstance().getContextManager().getInstanceContext().getEventBusContext();
            ShardingSphereDatabase database = metaDataContexts.getMetaData().getDatabase(connectionSession.getDatabaseName());
            DatabaseType protocolType = database.getProtocolType();
            DatabaseType databaseType = database.getResource().getDatabaseType();
            ExecuteProcessEngine.initialize(queryContext, executionGroupContext, eventBusContext);
            SQLStatementContext<?> context = queryContext.getSqlStatementContext();
            List<ExecuteResult> result = jdbcExecutor.execute(executionGroupContext,
                    ProxyJDBCExecutorCallbackFactory.newInstance(type, protocolType, databaseType, context.getSqlStatement(), databaseCommunicationEngine, isReturnGeneratedKeys, isExceptionThrown,
                            true),
                    ProxyJDBCExecutorCallbackFactory.newInstance(type, protocolType, databaseType, context.getSqlStatement(), databaseCommunicationEngine, isReturnGeneratedKeys, isExceptionThrown,
                            false));
            ExecuteProcessEngine.finish(executionGroupContext.getExecutionID(), eventBusContext);
            return result;
        } finally {
            ExecuteProcessEngine.clean();
        }
    }

As shown above, ExecuteProcessEngine.initialize(queryContext, executionGroupContext, eventBusContext); will store the SQL information in the two cache Maps. Finally, ExecuteProcessEngine.clean(); in the code block will clear up the Map in the cache.

The SQL shown in the Show processlist was obtained from processContexts. But this Map is just a local cache. If ShardingSphere is deployed in cluster mode, how does Show processlist obtain SQL running on other machines in the cluster? Let's see how ShardingSphere handles it.

2.2 How does `Show processlist` work?

When ShardingSphere receives the Show process command, it is sent to the executor ShowProcessListExecutor#execute for processing. The implementation of the getQueryResult() is the focus.

public final class ShowProcessListExecutor implements DatabaseAdminQueryExecutor {

    private Collection<String> batchProcessContexts;

    @Getter
    private QueryResultMetaData queryResultMetaData;

    @Getter
    private MergedResult mergedResult;

    public ShowProcessListExecutor() {
        ProxyContext.getInstance().getContextManager().getInstanceContext().getEventBusContext().register(this);
    }

    @Subscribe
    public void receiveProcessListData(final ShowProcessListResponseEvent event) {
        batchProcessContexts = event.getBatchProcessContexts();
    }

    @Override
    public void execute(final ConnectionSession connectionSession) {
        queryResultMetaData = createQueryResultMetaData();
        mergedResult = new TransparentMergedResult(getQueryResult());
    }

    private QueryResult getQueryResult() {
        ProxyContext.getInstance().getContextManager().getInstanceContext().getEventBusContext().post(new ShowProcessListRequestEvent());
        if (null == batchProcessContexts || batchProcessContexts.isEmpty()) {
            return new RawMemoryQueryResult(queryResultMetaData, Collections.emptyList());
        }
        Collection<YamlExecuteProcessContext> processContexts = new LinkedList<>();
        for (String each : batchProcessContexts) {
            processContexts.addAll(YamlEngine.unmarshal(each, BatchYamlExecuteProcessContext.class).getContexts());
        }
        List<MemoryQueryResultDataRow> rows = processContexts.stream().map(processContext -> {
            List<Object> rowValues = new ArrayList<>(8);
            rowValues.add(processContext.getExecutionID());
            rowValues.add(processContext.getUsername());
            rowValues.add(processContext.getHostname());
            rowValues.add(processContext.getDatabaseName());
            rowValues.add("Execute");
            rowValues.add(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis() - processContext.getStartTimeMillis()));
            int processDoneCount = processContext.getUnitStatuses().stream().map(each -> ExecuteProcessConstants.EXECUTE_STATUS_DONE == each.getStatus() ? 1 : 0).reduce(0, Integer::sum);
            String statePrefix = "Executing ";
            rowValues.add(statePrefix + processDoneCount + "/" + processContext.getUnitStatuses().size());
            String sql = processContext.getSql();
            if (null != sql && sql.length() > 100) {
                sql = sql.substring(0, 100);
            }
            rowValues.add(null != sql ? sql : "");
            return new MemoryQueryResultDataRow(rowValues);
        }).collect(Collectors.toList());
        return new RawMemoryQueryResult(queryResultMetaData, rows);
    }

    private QueryResultMetaData createQueryResultMetaData() {
        List<RawQueryResultColumnMetaData> columns = new ArrayList<>();
        columns.add(new RawQueryResultColumnMetaData("", "Id", "Id", Types.VARCHAR, "VARCHAR", 20, 0));
        columns.add(new RawQueryResultColumnMetaData("", "User", "User", Types.VARCHAR, "VARCHAR", 20, 0));
        columns.add(new RawQueryResultColumnMetaData("", "Host", "Host", Types.VARCHAR, "VARCHAR", 64, 0));
        columns.add(new RawQueryResultColumnMetaData("", "db", "db", Types.VARCHAR, "VARCHAR", 64, 0));
        columns.add(new RawQueryResultColumnMetaData("", "Command", "Command", Types.VARCHAR, "VARCHAR", 64, 0));
        columns.add(new RawQueryResultColumnMetaData("", "Time", "Time", Types.VARCHAR, "VARCHAR", 10, 0));
        columns.add(new RawQueryResultColumnMetaData("", "State", "State", Types.VARCHAR, "VARCHAR", 64, 0));
        columns.add(new RawQueryResultColumnMetaData("", "Info", "Info", Types.VARCHAR, "VARCHAR", 120, 0));
        return new RawQueryResultMetaData(columns);
    }
}

You’ll use the guava package's EventBus function, which is an information publish/subscribe database that is an elegant implementation of the Observer pattern. EventBus decouples classes from each other, and you'll find out more about it below.

getQueryResult() method will post ShowProcessListRequestEvent. ProcessRegistrySubscriber#loadShowProcessListData uses the @Subscribe annotations to subscribe to the event.

This method is the core to implementing Show processlist. Next, we'll introduce specific procedures of this method.

public final class ProcessRegistrySubscriber {    
    @Subscribe
    public void loadShowProcessListData(final ShowProcessListRequestEvent event) {
        String processListId = new UUID(ThreadLocalRandom.current().nextLong(), ThreadLocalRandom.current().nextLong()).toString().replace("-", "");
        boolean triggerIsComplete = false;
        // 1. Obtain the Process List path of all existing proxy nodes in cluster mode
        Collection<String> triggerPaths = getTriggerPaths(processListId);
        try {
            // 2. Iterate through the path and write an empty string to the node, to trigger the node monitoring.
            triggerPaths.forEach(each -> repository.persist(each, ""));
            // 3. Lock and wait 5 seconds for each node to write the information of currently running SQL to the persistence layer. 
            triggerIsComplete = waitAllNodeDataReady(processListId, triggerPaths);
            // 4. Fetch and aggregate the data written by each proxy node from the persistence layer. Then EventBus will post a ShowProcessListResponseEvent command, which means the operation is completed.
            sendShowProcessList(processListId);
        } finally {
            // 5. Delete resources
            repository.delete(ProcessNode.getProcessListIdPath(processListId));
            if (!triggerIsComplete) {
                triggerPaths.forEach(repository::delete);
            }
        }
    }
}

It contains five steps and steps 2 & 3 are the focus.

2.2.1 Step 2: the cluster obtains the data implementation

In this step, an empty string will be written to the node /nodes/compute_nodes/process_trigger/<instanceId>:<processlistId>, which will trigger ShardingSphere's monitoring logic.

When ShardingSphere is started, the persistence layer will watch to monitor a series of path changes, such as the addition, deletion, and modification operations of the path /nodes/compute_nodes.

However, monitoring is an asynchronous process and the main thread does not block, so step 3 is required to lock and wait for each ShardingSphere node to write its currently running SQL information into the persistence layer.

Let's take a look at how the ShardingSphere handles the monitoring logic.

public final class ComputeNodeStateChangedWatcher implements GovernanceWatcher<GovernanceEvent> {

    @Override
    public Collection<String> getWatchingKeys(final String databaseName) {
        return Collections.singleton(ComputeNode.getComputeNodePath());
    }

    @Override
    public Collection<Type> getWatchingTypes() {
        return Arrays.asList(Type.ADDED, Type.UPDATED, Type.DELETED);
    }

    @SuppressWarnings("unchecked")
    @Override
    public Optional<GovernanceEvent> createGovernanceEvent(final DataChangedEvent event) {
        String instanceId = ComputeNode.getInstanceIdByComputeNode(event.getKey());
        if (!Strings.isNullOrEmpty(instanceId)) {
            ...
        } else if (event.getKey().startsWith(ComputeNode.getOnlineInstanceNodePath())) {
            return createInstanceEvent(event);
            // show processlist
        } else if (event.getKey().startsWith(ComputeNode.getProcessTriggerNodePatch())) {
            return createShowProcessListTriggerEvent(event);
            // kill processlistId
        } else if (event.getKey().startsWith(ComputeNode.getProcessKillNodePatch())) {
            return createKillProcessListIdEvent(event);
        }
        return Optional.empty();
    }


    private Optional<GovernanceEvent> createShowProcessListTriggerEvent(final DataChangedEvent event) {
        Matcher matcher = getShowProcessTriggerMatcher(event);
        if (!matcher.find()) {
            return Optional.empty();
        }
        if (Type.ADDED == event.getType()) {
            return Optional.of(new ShowProcessListTriggerEvent(matcher.group(1), matcher.group(2)));
        }
        if (Type.DELETED == event.getType()) {
            return Optional.of(new ShowProcessListUnitCompleteEvent(matcher.group(2)));
        }
        return Optional.empty();
    }
}

After ComputeNodeStateChangedWatcher#createGovernanceEvent monitored the information, it would distinguish which event to create according to the path.

As shown in the above code, it is a new node, so ShowProcessListTriggerEvent will be posted. As each ShardingSphere instance will monitor /nodes/compute_nodes, each instance will process ShowProcessListTriggerEvent.

In this case, single-machine processing is transformed into cluster processing. Let's look at how ShardingSphere handles it.

public final class ClusterContextManagerCoordinator {
    @Subscribe
    public synchronized void triggerShowProcessList(final ShowProcessListTriggerEvent event) {
        if (!event.getInstanceId().equals(contextManager.getInstanceContext().getInstance().getMetaData().getId())) {
            return;
        }
        Collection<ExecuteProcessContext> processContexts = ShowProcessListManager.getInstance().getAllProcessContext();
        if (!processContexts.isEmpty()) {
            registryCenter.getRepository().persist(ProcessNode.getProcessListInstancePath(event.getProcessListId(), event.getInstanceId()),
                    YamlEngine.marshal(new BatchYamlExecuteProcessContext(processContexts)));
        }
        registryCenter.getRepository().delete(ComputeNode.getProcessTriggerInstanceIdNodePath(event.getInstanceId(), event.getProcessListId()));
    }
}

ClusterContextManagerCoordinator#triggerShowProcessList will subscribe to ShowProcessListTriggerEvent, in which processContext data is processed by itself. ShowProcessListManager.getInstance().getAllProcessContext() retrieves the processContext that is currently running (here the data refers to the SQL information that ShardingSphere stores in the Map before each SQL execution, which is described at the beginning of the article) and transfers it to the persistence layer. If the /nodes/compute_nodes/process_trigger/<instanceId>:<processlistId> node is deleted, the processing is completed.

When you delete the node, monitoring will also be triggered and ShowProcessListUnitCompleteEvent will be posted. This event will finally awake the pending lock.

public final class ClusterContextManagerCoordinator {

    @Subscribe
    public synchronized void completeUnitShowProcessList(final ShowProcessListUnitCompleteEvent event) {
        ShowProcessListSimpleLock simpleLock = ShowProcessListManager.getInstance().getLocks().get(event.getProcessListId());
        if (null != simpleLock) {
            simpleLock.doNotify();
        }
    }
}

2.2.2 Step 3: lock and wait for the data implementation

ShardingSphere uses the isReady(Paths) method to determine whether all instances have been processed. It returns true only when all instances have been processed.

There is a maximum waiting time of 5 seconds for data processing. If the processing is not completed in 5 seconds, then false is returned.

`public final class ClusterContextManagerCoordinator {

@Subscribe
public synchronized void completeUnitShowProcessList(final ShowProcessListUnitCompleteEvent event) {
    ShowProcessListSimpleLock simpleLock = ShowProcessListManager.getInstance().getLocks().get(event.getProcessListId());
    if (null != simpleLock) {
        simpleLock.doNotify();
    }
}

2.2.3 Aggregate the `processList` data and return it

After each instance processed the data, the instance that received the Show processlist command needs to aggregate the data and then display the result.

public final class ProcessRegistrySubscriber {  

    private void sendShowProcessList(final String processListId) {
        List<String> childrenKeys = repository.getChildrenKeys(ProcessNode.getProcessListIdPath(processListId));
        Collection<String> batchProcessContexts = new LinkedList<>();
        for (String each : childrenKeys) {
            batchProcessContexts.add(repository.get(ProcessNode.getProcessListInstancePath(processListId, each)));
        }
        eventBusContext.post(new ShowProcessListResponseEvent(batchProcessContexts));
    }
}

ProcessRegistrySubscriber#sendShowProcessList will aggregate the running SQL data into batchProcessContexts, and then post ShowProcessListResponseEvent.

This event will be consumed by ShowProcessListExecutor#receiveProcessListData, and the getQueryResult() method will proceed to show the queryResult.

So far, we’ve completed the execution process of Show processlist command.

2.3 How does Kill work?

Kill <processId> shares a similar logic with Show processlist, that is to combine EventBus with the watch mechanism.

Since we do not know which SQL the processId belongs to, it is also necessary to add empty nodes for each instance.

Through the watch mechanism, each ShardingSphere instance watches to the new node and checks whether the processId key is in the cache Map. If yes, fetch the value corresponding to the key.

The value is a Collection<Statement> collection. Then you only have to iterate through the Statement collection and call statement.cancel() in turn. The underlying layer is java.sql.Statement#cancel() method called to cancel SQL execution.

3. Conclusion

Currently, Apache ShardingSphere can only implement the Show processlist and Kill <processId> functions for MySQL dialects.

Once you get to know how they work, and if you’re interested, you‘re welcome to participate in the development of related functions. Our community is very open and anyone who is interested in contributing to open source code is welcome.

Relevant Links:

Apache ShardingSphere Official Website

Apache ShardingSphere GitHub

Apache ShardingSphere Slack Channel

Author

Xu Yang, a middleware R&D engineer at Servyou Group. Responsible for the table and database sharding with massive data. An open source enthusiast and ShardingSphere contributor. Currently, he’s interested in developing the kernel module of the ShardingSphere project.

Apache ShardingSphere 5.2.0 is Released!

Apache ShardingSphere — Tue, 13 Sep 2022 02:45:10 +0000

Our new 5.2.0 release enhances features such as SQL audit, elastic migration, SQL execution process management, and data governance on the cloud.

Introduction

Since Apache ShardingSphere released version 5.1.2 at the end of June, our community has continued to optimize and enhance its product features. The community merged 1,728 PRs from teams and individuals around the world. The resulting 5.2.0
release has been optimized in terms of its features, performance, testing, documentation, examples, etc.

The establishment of the shardingsphere-on-cloud sub-project shows ShardingSphere’s commitment to being cloud native. We welcome anyone interested in Go, database, and cloud to join the shardingsphere-on-cloud community.

The 5.2.0 release brings the following highlights:

SQL audit for data sharding.
Elastic data migration.
SQL execution process management.
Shardingsphere-on-cloud sub-project goes live. Newly added features, including SQL audit for data sharding and MySQL SHOW PROCESSLIST & KILL, can enhance users’ capability to manage ShardingSphere.

The SQL audit feature allows users to manage SQL audit to prevent the business system from being disrupted by inefficient SQL. The MySQL SHOW PROCESSLIST & KILL feature allows users to quickly view the SQL in execution through the SHOW PROCESSLIST statement, and forcibly cancel slow SQL.

The new version also supports elastic data migration. It supports the migration of data from Oracle, MySQL, and PostgreSQL to the distributed database ecosystem composed of ShardingSphere + MySQL or PostgreSQL, completing the transformation from a single database to a distributed one. The ShardingSphere community will support more features for heterogeneous database migration in future releases. Stay tuned for more updates.

The new version also transferred Helm Charts from the ShardingSphere repository to the shardingsphere-on-cloud sub-project. It is designed to provide distributed database solutions of ShardingSphere + MySQL or PostgreSQL on the cloud. This version significantly improves SQL parsing support for different databases and upgrades DistSQL’s parameter usage specifications. It removes the Memory operating mode from ShardingSphere mode and supports distributed transactions across multiple logical databases. This post will introduce the updates of ShardingSphere 5.2.0.

Highlights

SQL audit for data sharding

In large-scale data sharding scenarios, if a user executes an SQL query without the sharding feature, the SQL query will be routed to the underlying database for execution.

As a result, a large number of database connections will be occupied and businesses will be severely affected by timeout or other issues. If the user performs UPDATE/DELETE operations, a large amount of data may be incorrectly updated or deleted.

In response to the above problems, ShardingSphere 5.2.0 provides the SQL audit for data sharding feature and allows users to configure audit strategies. The strategy specifies multiple audit algorithms, and users can decide whether audit rules should be disabled. If any audit algorithm fails to pass, SQL execution will be prohibited. The configuration of SQL audit for data sharding is as follows.

rules:
- !SHARDING
  tables:
    t_order:
      actualDataNodes: ds_${0..1}.t_order_${0..1}
      tableStrategy:
        standard:
          shardingColumn: order_id
          shardingAlgorithmName: t_order_inline
      auditStrategy:
        auditorNames:
          - sharding_key_required_auditor
        allowHintDisable: true
  defaultAuditStrategy:
    auditorNames:
      - sharding_key_required_auditor
    allowHintDisable: true
  auditors:
    sharding_key_required_auditor:
      type: DML_SHARDING_CONDITIONS

In view of complex business scenarios, the new feature allows users to dynamically disable the audit algorithm by using SQL hints so that the business SQL that is allowable in partial scenarios can be executed. Currently, ShardingSphere 5.2.0 has a built-in DML disables full-route audit algorithm. Users can also implement ShardingAuditAlgorithm interface by themselves to realize more advanced SQL audit functions.

/* ShardingSphere hint: disableAuditNames=sharding_key_required_auditor */ SELECT * FROM t_order;

Elastic data migration

Data migration has always been a focus of the ShardingSphere community. Before 5.2.0, users needed to add an external table as a single sharding table, and then modify the sharding rules to trigger the migration, which was too complex and difficult for ordinary users.

To improve the ease of data migration, ShardingSphere 5.2.0 provides a new data migration feature, coupled with DistSQL for elastic migration. Users can migrate data from the existing single database to the distributed database system composed of ShardingSphere + MySQL or PostgreSQL in an SQL-like manner, achieving the transformation from a single database to a distributed one.

The new feature is capable of migrating Oracle data to PostgreSQL. Users can create sharding rules and sharding tables through DistSQL first, that is to create new distributed databases and tables, and then run MIGRATE TABLE ds.schema.table INTO table to trigger data migration.

During the migration process, users can also use the dedicated DistSQL for data migration in the table to manage the migration job status and data consistency. For more information about the new feature, please refer to the official document [Data Migration].

SQL Execution Process Management

The native MySQL database provides the SHOW PROCESSLIST statement, allowing the user to view the currently running thread. Users can kill the thread with the KILL statement for SQL that takes too long to be temporarily terminated.

The SHOW PROCESSLIST and KILL statements are widely used in daily operation and maintenance management. To enhance users’ ability to manage ShardingSphere, version 5.2.0 supports the MySQL SHOW PROCESSLIST and KILL statements. When a user executes a DDL/DML statement through ShardingSphere, ShardingSphere automatically generates a unique UUID identifier as an ID and stores the SQL execution information in each instance.

The following figure shows the results while executing the SHOW PROCESSLIST and KILL statements in ShardingSphere. When the user executes the SHOW PROCESSLIST statement, ShardingSphere processes the SQL execution information based on the current operating mode.

If the current mode is cluster mode, ShardingSphere collects and synchronizes the SQL execution information of each compute node through the governance center, and then returns the summary to the user. If the current mode is the standalone mode, ShardingSphere only returns SQL execution information in the current compute node.

The user determines whether to execute the KILL statement based on the result returned by the SHOW PROCESSLIST, and ShardingSphere cancels the SQL in execution based on the ID in the KILL statement.

Shardingsphere-on-cloud sub-project goes live

Shardingsphere-on-cloud is a project of Apache ShardingSphere providing cloud-oriented solutions. Version 0.1.0 has been released and it has been officially voted as a sub-project of Apache ShardingSphere.

Shardinsphere-on-cloud will continue to release various configuration templates, deployment scripts, and other automation tools for ShardingSphere on the cloud.

It will also polish the engineering practices in terms of high availability, data migration, observability, shadow DB, security, and audit, optimize the delivery mode of Helm Charts, and continue to enhance its cloud native management capabilities through Kubernetes Operator. Currently, there are already introductory issues in the project repository to help those who are interested in Go, Database, and Cloud to quickly get up and running.

Enhancement

Kernel

The ShardingSphere community optimizes the SQL parsing capability of different databases in this release, greatly improving ShardingSphere’s SQL compatibility.

Detailed SQL parsing optimization can be seen in the update log section below. It’s a long-term mission for the ShardingSphere community to improve SQL parsing support. Anyone who is interested is welcome to work with us.

Version 5.2.0 also supports the column-visible feature for MySQL, Oracle, SQLServer, and H2 databases, in a bid to meet the requirements of business SQL compatibility during a system upgrade. The read/write splitting feature supports the Cartesian product configuration, which greatly simplifies user configurations.

Access Port

In version 5.2.0, ShardingSphere-Proxy is capable of monitoring specified IP addresses and integrates openGauss database drivers by default. ShardingSphere-JDBC supports c3p0 data sources, and Connection.prepareStatement can specify the columns.

Distributed Transaction

In terms of distributed transactions, the original logical database-level transaction manager is adjusted to a global manager, supporting distributed transactions across multiple logical databases.

At the same time, it removed the XA statement’s ability to control distributed transactions as XA transactions are now automatically managed by ShardingSphere, which simplifies the operation for users.

Update logs

Below are all the update logs of ShardingSphere 5.2.0. To deliver a better user experience, this release adjusted the API of part of the functions, which can be seen from the API changes part below.

New Feature

Kernel: Support SQL audit for sharding feature
Kernel: Support MySQL show processlist and kill process list id feature
Scaling: Add dedicated DistSQL for data migration
Scaling: Basic support for migration of data to heterogeneous database
DistSQL: New syntax CREATE/ALTER/SHOW MIGRATION PROCESS CONFIGURATION
DistSQL: New syntax ALTER MIGRATION PROCESS CONFIGURATION
DistSQL: New syntax SHOW MIGRATION PROCESS CONFIGURATION
DistSQL: New syntax ADD MIGRATION SOURCE RESOURCE
DistSQL: New syntax SHOW SQL_TRANSLATOR RULE
DistSQL: New syntax CREATE SHARDING AUDITOR
DistSQL: New syntax ALTER SHARDING AUDITOR
DistSQL: New syntax SHOW SHARDING AUDIT ALGORITHMS

Enhancement

Kernel: Support column visible feature for MySQL, Oracle, SQLServer and H2
Kernel: Support cartesian product configuration for read/write splitting
Kernel: Support spring namespace and spring boot usage for sql translator
Kernel: Support JSR-310 Year and Month in IntervalShardingAlgorithm
Kernel: Support broadcast table update/delete limit statement
Kernel: Support create index on table(column) statement rewrite when config encrypts
Kernel: Support openGauss cursor, fetch, move, close statement for sharding and read/write splitting
Kernel: Support encrypt column rewrite when execute column is null in predicate
Kernel: Support encrypt show create table return logic columns
Kernel: Support create table with index statement rewrite when config encrypt
Kernel: Support PostgreSQL create operator statement parse
Kernel: Support PostgreSQL create materialized view statement parse
Kernel: Support PostgreSQL nested comments parse
Kernel: Support PostgreSQL alter subscription statement parse
Kernel: Support PostgreSQL create group statement parse
Kernel: Support PostgreSQL alter statictics statement parse
Kernel: Support PostgreSQL create foreign table statement parse
Kernel: Support PostgreSQL alter server statement parse
Kernel: Support PostgreSQL create foreign data wrapper statement parse
Kernel: Support PostgreSQL create event trigger statement parse
Kernel: Support PostgreSQL security label statement parse
Kernel: Support PostgreSQL reindex statement parse
Kernel: Support PostgreSQL reassign owned statement and refresh materialized view statement parse
Kernel: Support PostgreSQL prepare transaction statement parse
Kernel: Support PostgreSQL create collation statement parse
Kernel: Support PostgreSQL lock statement parse
Kernel: Support PostgreSQL alter rule statement parse
Kernel: Support PostgreSQL notify statement parse
Kernel: Support PostgreSQL unlisten statement parse
Kernel: Support Oracle alter function and alter hierarchy statement parse
Kernel: Support Oracle alter pluggable database statement parse
Kernel: Support Oracle alter materialized view log statement parse
Kernel: Support Oracle alter diskgroup statement parse
Kernel: Support Oracle alter operator statement parse
Kernel: Support oracle alter cluster statement parse
Kernel: Support oracle alter audit policy statement parse
Kernel: Support Oracle alter index type statement parse
Kernel: Support Oracle lock table statement parse
Kernel: Support Oracle alter java statement parse
Kernel: Support Oracle inline constraint statement parse
Kernel: Support openGauss geometric operator statement parse
Kernel: Optimize MySQL visible/invisible parse of create/alter table statements
Kernel: Support scope of variable prefixed with @@ in MySQL SET statement parse
Kernel: Support MySQL create procedure with create view parse
Kernel: Support column segments parse in create index on table statement
Kernel: Support openGauss cursor, fetch, move, close statement for sharding, readwrite-splitting
Kernel: Support encrypt column rewrite when execute column is null in predicate
Kernel: Support encrypt show create table return logic columns
Kernel: Support create table with index statement rewrite when config encrypt
Kernel: Support parsing ALTER LOCKDOWN PROFILE in Oracle
Kernel: Support parsing ALTER MATERIALIZED VIEW in Oracle
Kernel: Support parsing ALTER MATERIALIZED ZONEMAP in Oracle
Kernel: Support parsing ALTER LIBRARY in Oracle
Kernel: Support parsing ALTER INMEMORY JOIN GROUP in Oracle
Kernel: Support parsing DROP OPERATOR in Oracle
Kernel: Support parsing DROP RESTORE POINT in Oracle
Kernel: Support parsing CREATE RESTORE POINT in Oracle
Kernel: Support parsing DROP INMEMORY JOIN GROUP in Oracle
Kernel: Support parsing create_bit_xor_table in MySQL
Kernel: Support parsing MySQL DO statement
Kernel: Support parsing DropServer in openGauss
Kernel: Support parsing CREATE AGGREGATE In openGauss
Kernel: Support parsing ALTER ROUTINE in PostgreSQL
Kernel: Add PostgreSQL Create Cast Statement
Kernel: Add PostgreSQL Create Aggregate Statement
Kernel: Support fetch/move/close cursor statement in PostgreSQL
Kernel: Support Parsing ALTER PUBLICATION in PostgreSQL
Kernel: Add PostgreSQL Create Access Method Statement
Kernel: Support Parsing ALTER POLICY in PostgreSQL
Kernel: Support parsing ALTER OPERATOR in PostgreSQL
Kernel: Add PostgreSQL Copy Statement
Kernel: Add PostgreSQL Comment Statement
Kernel: Support listen statement in postgreSQL
Kernel: Support DECLARE cursor statement
Access port: Add default serverConfig in helm charts
Access port: Assemble openGauss JDBC Driver into Proxy distribution
Access port: ShardingSphere-Proxy listen on specified IP addresses
Access port: Support COM_STMT_SEND_LONG_DATA in MySQL Proxy
Access port: SELECT VERSION() support alias in MySQL Proxy
Access port: Fix openGauss Proxy could not be connected if no resource defined
Access port: Support using JRE defined in JAVA_HOME in ShardingSphere-Proxy’s startup script
Access port: Avoid client blocked when OOM occurred in ShardingSphere-Proxy
Access port: Support using c3p0 in ShardingSphere-JDBC
Access port: Support SET NAMES with value quoted by double-quote
Access port: Connection.prepareStatement with columns arguments is available in ShardingSphere-JDBC
Scaling: Improve MySQL connect and reconnect
Scaling: Fix MySQL json column may cause leak at incremental task
Scaling: Add permission check for PostgreSQL data sources
Scaling: Incremental migration support for MySQL MGR mode
Scaling: Improve job progress persistence
Scaling: Start job DistSQL execute and return synchronously
Scaling: Inventory migration support table has primary key and unique key
Scaling: Close unerlying ElasticJob when stopping job
Scaling: Improve logical replication slot name generation for PostgreSQL and openGauss
Scaling: Make query DistSQL could be executed when no database selected
DistSQL: Add worker_id to the result set of SHOW INSTANCE LIST & SHOW INSTANCE INFO
DistSQL: Improve the result of EXPORT DATABASE CONFIG
DistSQL: Support more databases for FORMAT SQL
DistSQL: Optimize the execution logic of CREATE TRAFFIC RULE
DistSQL: Add paramter writeDataSourceQueryEnabled for RDL READWRITE_SPLITTING RULE.
DistSQL: Support assistEncryptor for Encrypt RDL
DistSQL: Add sharding algorithm type check when CREATE SHARDING TABLE RULE
Distributed governance: Support database discovery to configure multiple groups of high availability under the same logic database
Distributed governance: Support ShardingSphere-Proxy to start up under empty logic library
Distributed governance: Support for isolating EventBus events by instance
Distributed governance: Support the database to detect changes in the master node and restart the detection heartbeat task
Distributed governance: Support ShardingSphere-Proxy to generate new worker-id when re-registering in cluster mode
Distributed governance: Thrown exception when inserting expression value in shadow column on executing insert
Distributed governance: Support distributed transactions across multiple logical databases
Distributed governance: Support executing truncate in XA & PostgreSQL
Distributed governance: Support alter local transaction rule with DistSQL
Distributed governance: Support global transaction manager
Distributed governance: Delete support for branch transaction on proxy

Bug Fix

Kernel: Fix single table metadata refresh error caused by filtering DataSourceContainedRule
Kernel: Fix parsing exception caused by the null value of MySQL blob type
Kernel: Fix PostgreSQL/openGauss reset statement parse error
Kernel: Fix wrong parameter rewrite when use sharding and encrypt
Kernel: Fix the failed conversion of Month related classes on IntervalShardingAlgorithm
Kernel: Fix NullPointerException when execute select union statement contains subquery
Kernel: Fix wrong encrypt rewrite result due to incorrect order of metadata
Kernel: Fix MySQL trim function parse error
Kernel: Fix MySQL insert values with _binary parse error
Access port: Fix MySQL syntax error cannot be thrown to client
Access port: Avoid EventLoop blocked because of closing JDBC resources
Access port: Correct server status flags returned by MySQL Proxy
Access port: Fix a possible connection leak issue if Proxy client disconnected in transaction
Access port: Fixed a possible consistency issue with the statement being executed when the Proxy client is disconnected
Access port: Avoid pooled connection polluted by executing SET statements
Access port: Make SHOW TABLES FROM work in ShardingSphere-Proxy
Access port: Fix PostgreSQL DDL could not be executed by Extended Query
Access port: Fix SHOW VARIABLES could not be executed in PostgreSQL Proxy without resource
Access port: Fix FileNotFoundException when use ShardingSphere Driver with SpringBoot fatjar
Scaling: Fix the problem that the table contains both primary key and unique index at inventory migration
Scaling: Improve incremental migration, support the latest position in the middle of batch insert event
Scaling: Fix the error caused by null field value in openGauss incremental migration
DistSQL: Fix incorrect strategy name in result of SHOW SHARDING TABLE RULES
DistSQL: Fix current rule config is modified in advance when ALTER SHARDING TABLE RULE
DistSQL: Fix connection leak when ALTER RESOURCE
DistSQL: Fix CREATE TRAFFIC RULE failed when load balance algorithm is null
Distributed governance: Fix that the monitoring heartbeat task was not stopped when the database was discovered and the logical library was deleted
Distributed governance: Fix cluster mode ShardingSphere-JDBC load all logic database
Distributed governance: Fix worker-id generated by SnowflakeKeyGenerateAlgorithm in cluster mode may exceed the maximum value
Shadow DB: Fix DistSQL adding shadow algorithm exception without shadow data source
Distributed transaction: Fix cross-database data source confusion caused by same data source name in multiple logical databases
Distributed transaction: Fix RUL DistSQL execution failure in transaction
Distributed transaction: Fix begin for PostgreSQL & openGauss
Agent: Fixed the error of null value in contextManager when collecting metric data

Refactor

Kernel: ShardingSphere metadata refactoring for splitting actual metadata and logical metadata
Kernel: Use ConnectionContext, QueryContext to remove ThreadLocal in FetchOrderByValueQueuesHolder, TrafficContextHolder, SQLStatementDatabaseHolder and TransactionHolder
Access port: Modify the default value of the ShardingSphere-Proxy version in the helm chart
Access port: Docker container will exit if ShardingSphere-Proxy failed to startup
Access port: Helm Charts in ShardingSphere repository are transferred to sub-project shardingsphere-on-cloud
Scaling: Plenty of refactor for better code reuse
DistSQL: Add a new category named RUL
Distributed governance: Refactor the schedule module and split it into cluster schedule and standalone schedule
Distributed governance: Remove memory mode, keep standalone mode and cluster mode
Distributed governance: Refactoring metadata table loading logic and persistence logic
Distributed governance: Refactoring distributed locks to retain the most concise interface design
Testing: Refactor: Unify The Creation for Proxy Containers in IT from ENV Modules
Testing: Refactor: Unify The Configuration for container created by testcontainer

API Changes

Kernel: Remove SQL passthrough to data source feature
Kernel: Add new assistedQueryEncryptorName and remove QueryAssistedEncryptAlgorithm interface
Kernel: Refactor readwrite-splitting api to improve user experience
Kernel: Remove check-duplicate-table-enabled configuration
Kernel: Remove useless config item show-process-list-enabled configuration
Scaling: Change keyword for part of data migration DistSQL
Scaling: Redesign part of data migration DistSQL
DistSQL: Unify parameter type specification
DistSQL: Split SHOW INSTANCE MODE to SHOW MODE INFO and SHOW INSTANCE INFO
DistSQL: Change DROP SCALING jobId to CLEAN MIGRATION jobId
DistSQL: Remove COUNT INSTANCE RULES
Distributed governance: Add database found that high availability supports all the slave libraries to go offline, and the main library undertakes the read traffic configuration
Distributed governance: SnowflakeKeyGenerateAlgorithm supports configuring worker-id in standalone mode
Shadow DB: Replace sourceDataSourceName with productionDataSourceName in Shadow API Configuration
Authority: Remove deprecated native authority provider

Relevant Links

🔗 Download Link

🔗 Update Logs

🔗 Project Address

🔗 Cloud Sub-project Address

Community Contribution

The Apache ShardingSphere 5.2.0 release is the result of 1,728 merged PRs, committed by 64 Contributors. Thank you for your efforts!

Author

Duan Zhengqiang, a senior middleware development engineer at SphereEx & Apache ShardingSphere PMC.

He started to contribute to Apache ShardingSphere middleware in 2018 and used to play a leading role in sharding practices dealing with massive data. With rich practical experience, he loves open-source and is willing to contribute. Now he focuses on the development of Apache ShardingSphere kernel module.

Apache ShardingSphere Enterprise User Case — Energy Monster

Apache ShardingSphere — Tue, 06 Sep 2022 09:39:41 +0000

Energy Monster’s application of ShardingSphere-JDBC

Energy Monster is a consumer tech company
with the mission to energize everyday life. The company is the largest provider of mobile device charging services in Asia.

As the company’s business concurrency volume is getting larger, the amount of data generated (users, orders, activities, etc.) increases each day. The traditional relational database has proven to be inadequate in supporting millions or tens of millions of data volumes in a single database or table.

Performance has been unable to meet the benchmark requirements of business development. Under these circumstances, data sharding is an effective way to solve the problem.

Technology selection

Under the Database Plus concept, ShardingSphere is designed to build an ecosystem on top of heterogeneous databases. The goal is to provide globally scalable and enhanced computing capabilities while maximizing the original database computing capabilities.

The interaction between applications and databases becomes oriented towards the Database Plus standard, therefore minimizing the impact of database fragmentation on upper-layer services.

Within the ShardingSphere ecosystem, ShardingSphere-JDBC is positioned as a lightweight Java framework, providing additional services in Java’s JDBC layer.

It uses the client to directly connect to the database and provide services in the form of a jar package, without additional deployment and dependence. It can be understood as an enhanced version of the JDBC driver, which is fully compatible with JDBC and various ORM frameworks.

ShardingSphere-JDBC enables developers to focus only on the work outside the data layer by coordinating the data read and write under the data sharding, instead of using business code to manually select databases and tables.

Business case

UCS is Energy Monster’s user-centric service providing basic functionality for users on the Server side. In 2018, it was stripped from PHP Server and moved to the Java technology stack to implement microservitization.

It involves the design of new databases and tables and data cleaning and migration. The whole switchover process was expected to ensure the following functions:

Stability: smooth release in a short time without halting.
Accuracy: ensure accurate cleaning of tens of millions of data volumes.
Scalability: solve the performance problems caused by increasing data volume and ensure scalability.

Solutions to data cleansing and migration

Initial data synchronization.
The application’s server cuts off the entry (users).
Data synchronization (updates and new users since the last time point).
Data cleaning.
User center release.

Data sharding strategy

The database adopts a database shards design, divided into 16 databases. The default shard key is user_id and the default sharding strategy user_id is mod 16, such as ${user_id % 16} for the user table. For SQL that does not carry shard keys, broadcast routing is used.

user_id is used as the shard key because user_id can cover most business scenarios, and other fields possibly can be empty. In the local test, the query of shard key strategy (openId,mobile) took 50ms to 200ms.

Using the sharding algorithm

There are currently three sharding algorithms available.

Standard sharding algorithm. It corresponds to StandardShardingAlgorithm, used for scenarios that use a single key as the shard key, such as =, IN, BETWEEN AND, >, <, > =, < =.
Complex sharding algorithm. It corresponds to ComplexKeysShardingAlgorithm, used for scenarios that use multi-key as the shard key. The logic with multiple shard keys is complex and requires developers to handle it by themselves.
Hint sharding algorithm. It corresponds to HintShardingAlgorithm, used for scenarios where the Hint row is used for sharding.

Upgrading ShardingSphere-JDBC

ShardingSphere-JDBC is used in multiple business scenarios, such as order, inventory, and finance. By 2021, the R&D groups or teams were using different versions of ShardingSphere-JDBC, ranging from 1.X to 4.X, which is difficult to achieve unified maintenance in the later stage.

Additionally, there are some potential bugs and missing functions in the earlier version. Based on requirements for unified management and availability, we implemented a project to unify the ShardingSphere-JDBC’s versions used by the company and upgrade them to a 4.1.1 stable version in April 2021.

The following problems were encountered during the upgrade:

1. It takes a long time to start the service after the upgrade.

ShardingSphere-JDBC checks the metadata consistency of sub-tables when the service is started. The configuration item max.connections.size.per.quer (maximum number of connections that can be opened per query) is 1 by default. With a large number of tables, the loading process would be slow. You need to refer to the connection pool configuration to improve the loading speed.

2. There is no response when there is no shard key in the sub-table query.

Logical SQL query does not specify shard keys and it queries all the tables according to the whole database tables router in broadcasting routing.

The configuration items have 108 pieces of real tables in a database. According to the configuration of maxConnectionsizeperquery=50, ShardingSphere-JDBC uses the connection limit mode, divides the query requests into three groups, and merges the results with in-memory. As a result, 36 database connections are required for one query. But the maxActive configured by the druid thread pool is set to 20, resulting in a deadlock.

Solutions:

Combine check.table.metadata.enabled=true（check the metadata consistency in sub-tables when started）and properly configure - maxConnectionSizePerQuery（maximum number of connections that can be opened by each query).
maxConnectionSizePerQuery should be less than the maximum number of active threads configured by the druid thread pool.

3. After upgrading from 1.X, an error message “Cannot update Sharding key” is displayed in SQL execution, and the actual shard key value is not updated.

To avoid data query failure caused by changing the shard key value, shard key detection is added to the SQL update in the 4.X version. The error can be rectified in the following ways:

remove the shard key when updating.
the shard key is added to the where statement synchronously.

4. A start failure is caused when using druid-spring-boot-starter, which is incompatible with Sharding-datasource.

The druid data connection pool starter will load and create a default data source. This will cause conflicts when ShardingSphere-JDBC creates data sources.

5. inline strategy reports an error in range query.

The inline strategy doesn't support range query by default and the standard strategy is advised. Add the following configuration if the inline strategy is needed for the range query.

spring.shardingsphere.props.allow.range.query.with.inline.sharding: true

Note: Here all the inline strategy range queries will query each sub-table in broadcasting.

6. The “Cannot find owner from table” error is reported.

SQL (simplified):

select id from (select id from x) as a group by a.id

The 4.X version supports limited sub-queries. This problem is caused by the name of the intermediate table. Remove the table alias of select or group order or other fields.

https://github.com/apache/shardingsphere/issues/4810

7. The table’s primary key conflicts with the primary key generated by the SNOWFLAKE algorithm.

ShardingSphere provides flexible ways to configure distributed primary key generation strategies. In the sharding rule configuration module, you can configure the primary key generation strategy for each table.

By default, the snowflake algorithm is used to generate long integer data of 64bit. The snowflake generator needs to be configured with:

spring.shardingsphere.sharding.tables.x.key-generator.props.worker.id = ${dcc.node.id}

The company uses the apollo configuration center to deliver the node id of the service instance. The service uses multi-data sources. If you use the YAML file to load sharding configuration, the workId cannot be automatically loaded into sharding configuration items.

Solutions:

Use the custom generator type based on the built-in SnowflakeShardingKeyGenerator.

If the primary key is used as a shard key, configure max.vibration.offset based on the data sharding value to increase the vibration range.

8. The 3.X version reports an error when CASE WHEN statement is executed.

First, the 3.X and 4.X versions don’t support the case when statement.

The 3.X and 4.X versions have different logics when parsing the shard keys of case when's update statement. The 4.X parserEngine.parse method will ignore the case when parsing parameters, resulting in inconsistency with the external parameter list and an error when 3.X executes the normal SQL.

The 3.X version works correctly because the first parameter of case when is intentionally set to the shard key when the SQL is written, and the case when statement comes first.

https://github.com/apache/shardingsphere/issues/13233

Solutions：

It is suggested to rewrite SQL as the case when is not supported.
According to the shard key parsing logic in version 4.1.1, case when is placed at the end, and the shard key remains the first parameter of case when.

9. The logical table actualDataNodes is configured and no default value error is reported for the primary key.

The check.table.metadata.enabled=true is not configured for service, and the metadata consistency of sub-tables is not checked by default.

The first table of actualDataNodes configured by services does not exist, resulting in an empty GenerateKeyContenxt.

Solutions：

Configure check.table.metadata.enabled=true. A non-existent table is detected when started and an error is reported.
Rewrite the actualDataNodes inline expression to make sure that the first table exists.

10. In version 3.0, there is a deadlock under the high concurrency of the full database and table router.

ShardingSphere-JDBC uses local transactions by default. In local transactions, the database connection is obtained asynchronously. Under high concurrency, it is possible that all database connections cannot be obtained, resulting in a deadlock.

Conclusion

As a ShardingSphere core user, Energy Monster’s upgrade process also reflects some problems that community users may encounter in the application of ShardingSphere.

Currently, Apache ShardingSphere’s stable version has been updated to 5.1.2 and has been optimized in terms of its functions, performance, testing, documentation, and examples.

You can refer to Apache ShardingSphere’s official website for more information. If you have any questions or suggestions, you are also welcome to give feedback on Github. The community will actively respond and discuss.

Project Links:

ShardingSphere Github

ShardingSphere Twitter

ShardingSphere Slack

Contributor Guide

GitHub Issues

Contributor Guide